Chapter 1 — Foundations: System One AI & Decision Economics

You will build: a swappable decision client (Jev, adapter, or offline fake) plus a latency and cost harness.

Capstone increment: capstone-project/src/triage/jev_client.py

1.1 The one-sentence model: a smart if

Ordinary code branches on values it can compute:

if order.total > 100:
    apply_free_shipping(order)

It breaks down when the condition is a judgment: is this customer describing fraud? How urgent is it? Which team should get it? Before Jev you had three options:

Jev takes a state (text or JSON) and a set of typed questions, and returns one typed answer per question with probabilities. It doesn’t write text, call tools, or run a loop. It returns numbers, and your code branches on them.

OpenAI Decisions - an Jev counterpart from OpenAI describes requests as three parts model (jev-1.13.0 or gpt-6-luna), state and questions. OpenAI is called state as input.

Calibration means its probabilities match reality: when it says 0.8 it’s right about 80% of the time, because it was trained for that (RLCD).

RLCD (Reinforcement Learning for Calibrated Decisions) is TypeSafe’s post-training method. It rewards Jev for probabilities that match how often it’s actually right, not just for picking the right answer. TypeSafe has published only the name, so the exact recipe (likely log-loss or Brier-style scoring) is inferred.

Jev Use Cases

1.2 Where Jev sits in a system

The rule for the whole book is Code computes. LLMs create. Jev decides. Fresh state proves. Every design decision in later chapters comes back to it.

A quick way to decide who owns a piece of work:

1.3 Economics you can put in a design doc

Published figures (TypeSafe, Sep 2026): $0.042 per million input tokens, output free, typically 70–500 ms end to end (most around 100 ms, measured from US West). Rate limits during early access: 250k tokens/s, 1,200 requests/min.

Don’t reason about cost per token. Reason about cost per solved task. Here’s a calculator you can keep in the repo:

# economics.py
from __future__ import annotations

from dataclasses import dataclass

JEV_USD_PER_M_INPUT: float = 0.042


@dataclass(frozen=True, slots=True)
class Workload:
    name: str
    calls_per_month: int
    input_tokens_per_call: int
    llm_usd_per_m_input: float
    llm_usd_per_m_output: float
    llm_output_tokens_per_call: int


def jev_monthly_cost(w: Workload) -> float:
    return w.calls_per_month * w.input_tokens_per_call * JEV_USD_PER_M_INPUT / 1_000_000


def llm_monthly_cost(w: Workload) -> float:
    inp = w.calls_per_month * w.input_tokens_per_call * w.llm_usd_per_m_input
    out = w.calls_per_month * w.llm_output_tokens_per_call * w.llm_usd_per_m_output
    return (inp + out) / 1_000_000


def blended_savings(total_ai_bill: float, decision_share: float, new_cost_ratio: float) -> float:
    """total savings = decision share x savings on those calls.

    decision_share: fraction of the bill spent on classify/route/score/verify (0..1)
    new_cost_ratio: new path cost / old path cost for those calls (e.g. 0.05)
    """
    if not 0 <= decision_share <= 1 or not 0 <= new_cost_ratio <= 1:
        raise ValueError("shares and ratios must be within [0, 1]")
    return total_ai_bill * decision_share * (1 - new_cost_ratio)


if __name__ == "__main__":
    triage = Workload(
        name="fintech-triage",
        calls_per_month=3_000_000,
        input_tokens_per_call=450,        # payload + 3 questions
        llm_usd_per_m_input=0.15,         # small LLM with JSON output
        llm_usd_per_m_output=0.60,
        llm_output_tokens_per_call=60,
    )
    print(f"Jev : ${jev_monthly_cost(triage):>10,.2f}/month")
    print(f"LLM : ${llm_monthly_cost(triage):>10,.2f}/month")
    # The Deep Dive example: $10k bill, 60% decisions, new path costs 5% of old.
    print(f"Blended saving: ${blended_savings(10_000, 0.60, 0.05):,.0f}")  # $5,700

Jev : $     56.70/month
LLM : $    310.50/month
Blended saving: $5,700

Two points to take from this:

1. Savings are capped by the decision share. If only 10% of your bill is classification, making it free saves at most 10%.
2. The Jevons paradox. Jev is named after William Stanley Jevons: cheaper steam engines drove coal consumption up. When a judgment costs a hundred-thousandth of a cent, you stop sampling and start checking every message, candidate, tool call and handoff. The real payoff is decision density, meaning checks you couldn’t afford to run before.

1.4 Getting access: two paths

First call with curl

export TYPESAFE_API_KEY=your_key_here
curl -s https://api.typesafe.ai/v1/systemone \
  -H "Authorization: Bearer $TYPESAFE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "jev-latest",
    "state": {"message": "Someone used my card in another city. I did not make this payment."},
    "questions": {
      "is_fraud_suspected": {
        "type": "noul",
        "instructions": "Does `message` describe a transaction the customer did not authorise?"
      }
    }
  }'

{
  "model": "jev-1.13.0",
  "answers": { "is_fraud_suspected": { "type": "noul", "noul": 0.97 } },
  "usage": { "input_tokens": 302, "output_tokens": 26 }
}

(Numbers are illustrative. The shape is exact.) Always log the model field. jev-latest moves, and your thresholds are calibrated against one specific version.

Error codes you will actually see

1.5 Capstone increment: a swappable decision client

We’ll protect the rest of the codebase from vendor types. The engine talks to one small interface, DecisionClient, and gets back our own normalized result types. That buys us three things:

• switch Jev ↔ adapter ↔ offline fake by environment variable,
• unit tests with no network,
• one place to record latency, tokens and model version for every decision.

# capstone-project/src/triage/jev_client.py
"""Swappable decision client. The rest of the engine never imports a vendor SDK."""
from __future__ import annotations

import logging
import os
import time
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from typing import Any, Protocol, TypeAlias

log = logging.getLogger("triage.jev_client")

JSONState: TypeAlias = str | Mapping[str, Any] | list[Any]
QuestionSpec: TypeAlias = Mapping[str, Any]          # {"type": "noul"|"choice"|"score", ...}
Questions: TypeAlias = Mapping[str, QuestionSpec]


# ---------- Normalised result types (vendor-neutral) ----------
@dataclass(frozen=True, slots=True)
class NoulResult:
    p_yes: float


@dataclass(frozen=True, slots=True)
class ChoiceResult:
    choice: str
    probabilities: Mapping[str, float]
    confidence: float

    def margin(self) -> float:
        """Gap between the top two options: a useful diagnostic next to `confidence`."""
        top = sorted(self.probabilities.values(), reverse=True)
        return top[0] - (top[1] if len(top) > 1 else 0.0)


@dataclass(frozen=True, slots=True)
class ScoreResult:
    score: float
    probabilities: Mapping[int, float]
    confidence: float
    levels: int

    def normalized(self) -> float:
        """Map 0..(levels-1) onto 0..1 so scores with different rubric lengths can be combined."""
        return self.score / (self.levels - 1) if self.levels > 1 else 0.0


Answer: TypeAlias = NoulResult | ChoiceResult | ScoreResult


@dataclass(frozen=True, slots=True)
class DecisionResponse:
    answers: Mapping[str, Answer]
    model: str
    input_tokens: int | None
    latency_ms: float
    backend: str
    raw: Any = field(default=None, repr=False, compare=False)

    def noul(self, qid: str) -> NoulResult:
        a = self.answers[qid]
        if not isinstance(a, NoulResult):
            raise TypeError(f"{qid} is {type(a).__name__}, expected NoulResult")
        return a

    def choice(self, qid: str) -> ChoiceResult:
        a = self.answers[qid]
        if not isinstance(a, ChoiceResult):
            raise TypeError(f"{qid} is {type(a).__name__}, expected ChoiceResult")
        return a

    def score(self, qid: str) -> ScoreResult:
        a = self.answers[qid]
        if not isinstance(a, ScoreResult):
            raise TypeError(f"{qid} is {type(a).__name__}, expected ScoreResult")
        return a


class DecisionError(RuntimeError):
    """Raised when no typed decision could be obtained. Callers must fall back."""

    def __init__(self, message: str, *, retryable: bool) -> None:
        super().__init__(message)
        self.retryable = retryable


class DecisionClient(Protocol):
    backend: str

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse: ...


# ---------- Conversion from any SDK-shaped response ----------
def _normalise(raw_answers: Mapping[str, Any], questions: Questions) -> dict[str, Answer]:
    out: dict[str, Answer] = {}
    for qid, spec in questions.items():
        if qid not in raw_answers:
            raise DecisionError(f"answer missing for question {qid!r}", retryable=False)
        a = raw_answers[qid]
        kind = spec["type"]
        if kind == "noul":
            out[qid] = NoulResult(p_yes=float(a.noul))
        elif kind == "choice":
            out[qid] = ChoiceResult(
                choice=str(a.choice),
                probabilities={str(k): float(v) for k, v in a.probabilities.items()},
                confidence=float(a.confidence),
            )
        elif kind == "score":
            out[qid] = ScoreResult(
                score=float(a.score),
                probabilities={int(k): float(v) for k, v in a.probabilities.items()},
                confidence=float(a.confidence),
                levels=len(spec["criteria"]),
            )
        else:  # pragma: no cover - guarded by contract validation
            raise DecisionError(f"unknown question type {kind!r}", retryable=False)
    return out


# ---------- Backend 1: TypeSafe Jev ----------
class JevDecisionClient:
    backend = "jev"

    def __init__(self, model: str | None = None, timeout_s: float = 2.0, max_retries: int = 2,
                 *, base_url: str | None = None, api_key: str | None = None) -> None:
        from typesafe_sdk import RetryPolicy, TypeSafeClient  # lazy: optional dependency

        self._model = model or os.getenv("JEV_MODEL", "jev-latest")
        self._client = TypeSafeClient(
            model=self._model,
            api_key=api_key,            # None -> TYPESAFE_API_KEY
            base_url=base_url,          # None -> TYPESAFE_BASE_URL or the hosted API (Ch9 points it at Kev)
            timeout=timeout_s,
            retry=RetryPolicy(max_retries=max_retries, timeout=timeout_s * (max_retries + 1)),
        )

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        from typesafe_sdk import (
            TypeSafeAPIConnectionError,
            TypeSafeAPIError,
            TypeSafeAuthenticationError,
            TypeSafeError,
            TypeSafeUnprocessableEntityError,
        )

        t0 = time.perf_counter()
        try:
            r = self._client.system_one(state=state, questions=dict(questions))
        except (TypeSafeAuthenticationError, TypeSafeUnprocessableEntityError) as e:
            raise DecisionError(f"non-retryable Jev error: {e}", retryable=False) from e
        except (TypeSafeAPIConnectionError, TypeSafeAPIError) as e:
            # SDK already retried 408/429/5xx with backoff; surface to the circuit breaker.
            raise DecisionError(f"Jev unavailable after retries: {e}", retryable=True) from e
        except TypeSafeError as e:
            raise DecisionError(f"Jev client error: {e}", retryable=False) from e
        latency_ms = (time.perf_counter() - t0) * 1000
        resp = DecisionResponse(
            answers=_normalise(r.answers, questions),
            model=r.model,
            input_tokens=r.usage.input_tokens,
            latency_ms=latency_ms,
            backend=self.backend,
            raw=r,
        )
        log.info("decision backend=%s model=%s tokens=%s latency_ms=%.1f",
                 self.backend, resp.model, resp.input_tokens, latency_ms)
        return resp

    def close(self) -> None:
        self._client.close()


# ---------- Backend 2: System One Adapter (build before access) ----------
class AdapterDecisionClient:
    """Same interface backed by a general LLM. NOT calibrated like Jev: never tune thresholds on it."""

    backend = "adapter"

    def __init__(self, provider: str = "openai", model: str = "gpt-4o-mini") -> None:
        from system_one_adapter import SystemOneAdapterClient

        self._provider, self._model = provider, model
        self._client = SystemOneAdapterClient(
            structured_outputs=True,
            llm_answer_mode="probabilities",
            normalize_probabilities=True,
        )

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        t0 = time.perf_counter()
        try:
            r = self._client.system_one(
                state=state, questions=dict(questions), provider=self._provider, model=self._model
            )
        except Exception as e:  # adapter wraps several provider SDKs
            raise DecisionError(f"adapter failure: {e}", retryable=True) from e
        return DecisionResponse(
            answers=_normalise(r.answers, questions),
            model=f"{self._provider}:{self._model}",
            input_tokens=None,
            latency_ms=(time.perf_counter() - t0) * 1000,
            backend=self.backend,
            raw=r,
        )


# ---------- Backend 3: offline fake for unit tests and demos ----------
FakeRule: TypeAlias = Callable[[JSONState, Questions], Mapping[str, Answer]]


class FakeDecisionClient:
    backend = "fake"

    def __init__(self, rule: FakeRule, latency_ms: float = 0.0, fail_times: int = 0) -> None:
        self._rule, self._latency_ms, self._fail_left = rule, latency_ms, fail_times

    def decide(self, state: JSONState, questions: Questions) -> DecisionResponse:
        if self._fail_left > 0:
            self._fail_left -= 1
            raise DecisionError("fake outage", retryable=True)
        if self._latency_ms:
            time.sleep(self._latency_ms / 1000)
        return DecisionResponse(
            answers=dict(self._rule(state, questions)),
            model="fake-0",
            input_tokens=None,
            latency_ms=self._latency_ms,
            backend=self.backend,
        )


def client_from_env() -> DecisionClient:
    backend = os.getenv("DECISION_BACKEND", "jev")
    if backend == "jev":
        return JevDecisionClient()
    if backend == "adapter":
        return AdapterDecisionClient(
            provider=os.getenv("ADAPTER_PROVIDER", "openai"),
            model=os.getenv("ADAPTER_MODEL", "gpt-4o-mini"),
        )
    raise ValueError(f"Unsupported DECISION_BACKEND={backend!r} (fake is for tests only)")

Why the questions are plain dicts: the SDK accepts question dictionaries ({"type": "noul", "instructions": ...}) as well as Choice/Score/Noul objects. Dicts are easy to version, diff, hash and store in a trace. Chapter 2 builds on that.

1.6 Mini-project 1: latency & cost harness

The harness runs the same contract N times, reports p50/p95/p99, and compares one fan-out call with N sequential single-question calls. Run it on day one and keep running it: numbers measured from Tirunelveli (a culturally rich city in Tamil Nadu, South India) won’t match numbers measured from San Francisco.

Note: p50/p95/p99 are latency percentiles: p50 is the typical (median) request time, while p95 and p99 are the times that 95% and 99% of requests finish within, which shows how slow the worst 1 in 20 and 1 in 100 calls get

# mini-projects/01-starter-client/bench.py
from __future__ import annotations

import statistics
import time
from collections.abc import Sequence
from dataclasses import dataclass

from triage.jev_client import DecisionClient, DecisionError, Questions, JSONState, client_from_env

JEV_USD_PER_M_INPUT = 0.042


@dataclass(frozen=True, slots=True)
class BenchResult:
    label: str
    n: int
    p50_ms: float
    p95_ms: float
    p99_ms: float
    errors: int
    usd_per_1k: float


def _pct(xs: Sequence[float], p: float) -> float:
    if not xs:
        return float("nan")
    s = sorted(xs)
    k = min(len(s) - 1, max(0, round(p / 100 * (len(s) - 1))))
    return s[k]


def bench(label: str, fn, n: int) -> BenchResult:
    lat: list[float] = []
    tokens: list[int] = []
    errors = 0
    for _ in range(n):
        t0 = time.perf_counter()
        try:
            used = fn()
        except DecisionError:
            errors += 1
            continue
        lat.append((time.perf_counter() - t0) * 1000)
        tokens.append(used)
    mean_tokens = statistics.fmean(tokens) if tokens else 0.0
    return BenchResult(label, n, _pct(lat, 50), _pct(lat, 95), _pct(lat, 99), errors,
                       mean_tokens * JEV_USD_PER_M_INPUT / 1_000_000 * 1000)


def fan_out(client: DecisionClient, state: JSONState, qs: Questions) -> int:
    r = client.decide(state, qs)
    return r.input_tokens or 0


def sequential(client: DecisionClient, state: JSONState, qs: Questions) -> int:
    total = 0
    for qid, spec in qs.items():
        total += client.decide(state, {qid: spec}).input_tokens or 0
    return total


if __name__ == "__main__":
    from triage.contract import TRIAGE_QUESTIONS  # built in Chapter 2

    client = client_from_env()
    state = {"report": "Two card payments of ₹ 49,999 I don't recognise, both at 3am. Please block my card."}
    for r in (
        bench("fan-out (1 call, 3 questions)", lambda: fan_out(client, state, TRIAGE_QUESTIONS), 50),
        bench("sequential (3 calls)", lambda: sequential(client, state, TRIAGE_QUESTIONS), 50),
    ):
        print(f"{r.label:32} p50={r.p50_ms:6.0f}ms p95={r.p95_ms:6.0f}ms "
              f"p99={r.p99_ms:6.0f}ms err={r.errors} ${r.usd_per_1k:.5f}/1k")

Output

Reference numbers to compare yours against (published, not ours):

Most of that saving comes from sending a big state once. With a small state the cost gap shrinks, but latency still favours fan-out. Concurrent separate calls close the latency gap but never the cost gap.

1.7 Checklist

• TYPESAFE_API_KEY in env / secret manager, never in state, logs or git
• JEV_MODEL pinned to a versioned id (e.g. jev-1.13.0) before any threshold tuning
• Every decision logs model, input_tokens, latency_ms, backend
• Vendor SDK imported in exactly one module (jev_client.py)
• Harness run from your region; p95 recorded in the design doc

What’s next

We have a client that can ask something. Chapter 2 is about what to ask: the three primitives, how to write questions Jev answers well, and the versioned decision contract for the triage engine (is_fraud_suspected, urgency_rating, dispatch_target) that the bench above already imports.