labs | 01 | typed call
lab 01 | ~8 min | masterclass

Every call returns a typed Result. Never a raw dict. Never None.

An agent that gets None back cannot tell "no results" apart from "auth died" or "rate limited." So it retries blindly, burns your quota, or silently drops data. The fix is a typed Result: a small object that carries both the answer and, on failure, a labeled reason. Pair it with an error taxonomy (a short, fixed list of failure names the caller can check). That is the difference between an agent that recovers and an agent that loops. It is the first discipline of the whole stack. Build it once here, and every later lab hands back this same shape.

step 1

Define the Result and the error taxonomy.

It is a small dataclass (a plain Python class that just holds fields) with four fields and two ways to build it. The taxonomy is a handful of named string constants, so a caller checks for a name like rate_limit, not a bare number like 429. Save as result.py.

result.py
from dataclasses import dataclass
from typing import Generic, Optional, TypeVar

T = TypeVar("T")

# The error taxonomy. A caller branches on one of these, never on a traceback.
AUTH_EXPIRED = "auth_expired"
RATE_LIMIT   = "rate_limit"
NOT_FOUND    = "not_found"
VALIDATION   = "validation"
TRANSIENT    = "transient"

@dataclass
class Result(Generic[T]):
    ok: bool
    value: Optional[T] = None
    error: Optional[str] = None        # human-readable detail
    error_kind: Optional[str] = None   # one taxonomy constant above

def ok_result(value):
    """A success. ok is True, value is set, error fields are None."""
    return Result(ok=True, value=value)

def err_result(kind, message):
    """A failure. ok is False, error_kind is a taxonomy constant."""
    return Result(ok=False, error=message, error_kind=kind)

if __name__ == "__main__":
    print(ok_result([1, 2, 3]))
    print(err_result(RATE_LIMIT, "429 from upstream"))

Two outcomes, one shape. ok_result(value) carries the answer. err_result(kind, message) carries one taxonomy name the caller can branch on. The fields never lie: if ok is True there is a value, and if it is False there is an error_kind. Run it with python result.py to see both printed.

step 2

Wrap a flaky call and classify the failure.

A real call can fail by raising an exception (Python's way of signaling an error mid-call). We catch it in one try/except, then run a classify_error helper that maps the exception to one taxonomy name. The client here is a tiny fake so the file runs on its own, but the wrapper is the exact shape you put around a database call, an HTTP client, or any vendor library. Add this to result.py below the two builders, or save it as fetch_user.py next to it.

fetch_user.py
from result import ok_result, err_result
from result import AUTH_EXPIRED, RATE_LIMIT, NOT_FOUND, TRANSIENT

class ClientError(Exception):
    """Stand-in for whatever your real client raises. Carries an HTTP status."""
    def __init__(self, status, message):
        super().__init__(message)
        self.status = status

# Tiny stub for the underlying client so this file runs on its own.
# Swap this for a real DB cursor or API call. The wrapper does not change.
_FAKE_DB = {"ray": {"id": "ray", "name": "Rayyan", "tier": "operator"}}

def _raw_get_user(uid):
    if uid == "expired":
        raise ClientError(401, "token expired")
    if uid == "flood":
        raise ClientError(429, "slow down")
    if uid not in _FAKE_DB:
        raise ClientError(404, "no such user: " + uid)
    return _FAKE_DB[uid]

def classify_error(exc):
    """Map a raised exception to one taxonomy constant. Status first, then fall through."""
    status = getattr(exc, "status", None)
    if status in (401, 403):
        return AUTH_EXPIRED
    if status == 429:
        return RATE_LIMIT
    if status == 404:
        return NOT_FOUND
    return TRANSIENT

def fetch_user(uid):
    """Return a Result. On success, value is the user dict. On failure, error_kind is set."""
    try:
        data = _raw_get_user(uid)
        return ok_result(data)
    except Exception as exc:
        return err_result(classify_error(exc), str(exc))

if __name__ == "__main__":
    for uid in ["ray", "expired", "flood", "ghost"]:
        r = fetch_user(uid)
        print(uid, "->", r.ok, r.error_kind or "(none)")

One try/except, one classifier. Every exit from fetch_user is a Result. The mapping uses HTTP status codes (the standard numbers a web server returns): 401 means unauthorized, 429 means too many requests, 404 means not found. So a 401 becomes auth_expired, a 429 becomes rate_limit, a 404 becomes not_found, and anything else becomes transient, where a retry is at least defensible. Run python fetch_user.py and watch all four paths return cleanly, with no error escaping the function.

step 3

The caller branches on error_kind, and logs one line.

This is the payoff. The caller never wraps fetch_user in its own try/except. It reads r.ok, then branches on r.error_kind to recover: re-authenticate on a dead token, back off on a throttle, skip-and-log on a missing record. It also writes one telemetry line per call (telemetry is just a record of what happened, for later inspection). The write is fire-and-forget, meaning if logging fails it is swallowed and never breaks the real call. This mirrors the real kernel's data/kernel_telemetry.jsonl. Save as caller.py and run it.

caller.py
import json, time, os
from fetch_user import fetch_user
from result import AUTH_EXPIRED, RATE_LIMIT, NOT_FOUND

TELEMETRY = "kernel_telemetry.jsonl"

def log(verb, ok, error_kind, ms):
    """Fire-and-forget: append one line, never raise into the caller."""
    try:
        line = {"verb": verb, "ok": ok, "error_kind": error_kind, "ms": ms}
        with open(TELEMETRY, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(line) + "\n")
    except Exception:
        pass  # telemetry must never break the call it is measuring

def reauth():        print("  -> re-authenticating, then retry once")
def backoff():       print("  -> backing off (exponential), then retry")
def log_and_skip(m): print("  -> skip and log: " + m)

def get_user(uid):
    t0 = time.time()
    r = fetch_user(uid)
    ms = int((time.time() - t0) * 1000)
    log("fetch_user", r.ok, r.error_kind, ms)

    if r.ok:
        print(uid, "OK ->", r.value["name"])
        return r.value

    # Branch on the taxonomy, not on a parsed traceback.
    if r.error_kind == AUTH_EXPIRED:
        reauth()
    elif r.error_kind == RATE_LIMIT:
        backoff()
    elif r.error_kind == NOT_FOUND:
        log_and_skip(r.error)
    else:
        log_and_skip("transient: " + str(r.error))
    return None

if __name__ == "__main__":
    for uid in ["ray", "expired", "flood", "ghost"]:
        print(uid, "...")
        get_user(uid)
    print("wrote telemetry to", os.path.abspath(TELEMETRY))
what you'll see (each branch fires off the one Result, telemetry on disk)
ray ...
ray OK -> Rayyan
expired ...
  -> re-authenticating, then retry once
flood ...
  -> backing off (exponential), then retry
ghost ...
  -> skip and log: no such user: ghost
wrote telemetry to ...\kernel_telemetry.jsonl

Four calls, four different recoveries, and zero try/except in the caller. The kernel_telemetry.jsonl file now holds one line per call with the verb, the outcome, the error_kind, and the latency (how many milliseconds it took). The real life kernel writes this same line on every call. That is what lets a dashboard show, for each verb, the success rate and the p95 latency (the time the slowest 5 percent of calls came in under), without anyone wiring up each call by hand.

why None is the most expensive return value you shiplife / kernel

None collapses three different worlds into one blank value. An empty result, dead auth, and being throttled all come back as the same None. A human reading the code can guess which one it is from context. An agent calling your function at 3am cannot. It will retry a failed login forever, because to it a dead token and an empty list look identical. Or it treats a rate-limit as "no data" and moves on, leaving a hole in its work that nobody notices until the report comes out wrong.

The typed error_kind is what lets a caller do the right thing on its own, with no human watching: re-authenticate on auth_expired, wait and retry on rate_limit, skip-and-log on not_found. The life kernel exists for this one reason. Callers branch on error_kind. They never read a stack trace, and they never guess.

agent, on a None return got None from fetch_user, unclear why. could be no user, could be auth, could be a throttle. retrying... retrying... retrying...
agent, on a typed Result error_kind=auth_expired. the token is dead, not the data. re-authenticating, then retrying once. [recovered]

Same failure, two completely different outcomes. The difference is one labeled field. Return the shape, and the machine on the other end can recover instead of loop.

hand this to your coding agent
Refactor every public function in <module>.py so it returns a typed Result
with .ok, .error, and .error_kind instead of raising or returning None.
Map exceptions to a taxonomy: auth_expired, rate_limit, not_found, validation,
transient. Add a classify_error() helper and a fire-and-forget telemetry line
per call. Then show me one caller rewritten to branch on error_kind.

Paste that into Claude Code or Cursor, point it at one real module, and read the diff. The agent does the mechanical work. You check that the taxonomy fits your actual ways of failing. That is the throughline of this whole masterclass: you write the discipline once, and your agent applies it everywhere.

checkpoint

Your code now tells the truth about failure. Any agent calling it can branch, retry, re-authenticate, or skip without reading a stack trace, and every call leaves a telemetry line behind. Every later lab in this masterclass hands back this same shape. So the recovery logic you just wrote is the one you will reuse on MCP (Model Context Protocol) tools, on remembered facts, and on a whole fleet of agents in the labs ahead.