Stateless in Practice

Stateless establishes the two channels an Effect[A, E, R] carries. A dependency is a Need that supply() answers. A failure is an exception that @throws lifts into the type and catch() takes back out. A type checker verifies that every caller either absorbs an Effect or declares it.

Every Ability so far comes from the library: the Need that supply() answers, and the Async that run() awaits. This chapter opens by writing an Ability from scratch, and that Ability turns out to be an ordinary class rather than a special form. The rest of the chapter adds:

• Handlers that make an unpredictable source testable
• A handler that swaps implementations while a program runs
• A cell of shared state that programs read and write through the channel
• A program whose signature is its own documentation, and whose body is only the success path
• A failure that enters the channel as a value, and a catch that takes the whole channel
• Dependency graphs that go deep, and a cast of Abilities that goes wide
• Decorators that add retry and parallelism to code they do not edit
• An account of what the guarantee does not cover

Abilities Are Not Special

A custom Ability is a request you design. A Need asks for an instance and gets whatever the handler supplies. Your own Ability can ask for anything you can name, and the handler answering it is an ordinary function, so the answer can differ at every request.

An Ability subclasses Ability[T], where T is the type its handler returns. Ability declares no __slots__, so you decorate an Ability with @record(slots=False). The base gives every instance its __dict__ back, and the flag says so at the class. Here is the Stateless version of Ask and Tell from Effect Management:

# ask_tell_stateless.py
from record import record
from stateless import Ability, Depend, handle, run

@record(slots=False)
class Ask(Ability[str]):
    prompt: str

@record(slots=False)
class Tell(Ability[None]):
    message: str

def ask(prompt: str) -> Depend[Ask, str]:
    answer: str = yield from Ask(prompt)
    return answer

def tell(message: str) -> Depend[Tell, None]:
    yield from Tell(message)

def greet() -> Depend[Ask | Tell, None]:
    name = yield from ask("What is your name? ")
    yield from tell(f"Hello, {name}!")

messages: list[str] = []

def capture(request: Tell) -> None:
    messages.append(request.message)

def scripted(request: Ask) -> str:
    return "Alice"

half = handle(capture)(greet)
full = handle(scripted)(half)
run(full())
print(messages)
#: ['Hello, Alice!']

Inside ask(), the operand of yield from is not a generator but the Ability object. yield from calls iter() on its operand, and Ability.__iter__ is a generator function, so the delegation target is an ordinary generator after all. Stripped of error handling, that method is:

def __iter__(self: Self) -> Generator[Self, T, T]:
    v = yield self
    return v

It yields once, so the handler receives one request. The yield from then evaluates to that generator’s return value, the return-channel rule, and here that value is whatever the handler sent back. prompt is payload on the request, there for the handler to read, so the answer to an Ask is whatever scripted() returns. A Tell needs no answer, which is why Tell is Ability[None] and capture() returns None.

ask() and tell() are accessors, small functions that each wrap one Ability and declare its answer type. need() has the same shape, and the ZIO listing in Effect Management has an accessor object doing the same job. The declared Depend[Ask, str] types name as str inside greet(). You can skip the accessor and write yield from Ask(...) inside greet(), and both the program and the type checker still work, because ty reads the answer type from Ask’s base, Ability[str]. A bare yield Ask(...) also runs, since the handler receives the same request, but the answer then has the type Any. A generator’s send type is all the checker knows about a bare yield, and an Effect’s send type is Any. With name typed Any, a mistake such as name + 1 passes the check. The accessor adds a name for the request and one place to state that answer type: the answer: str binding inside ask(), one line above the Depend[Ask, str] that repeats it to callers.

That annotation reads Depend[Ask, str], not Depend[Need[Ask], str], the distinction Waiting on a Coroutine draws for Async. Ask is an Ability, so the channel names it bare. The type bound makes the bare name more than a convention. The type checker rejects Depend[Console, None] at the annotation, before it examines any yield, because Console is not assignable to Ability[Any].

handle() reads the annotation on its argument to decide which Ability it answers, so scripted and capture must annotate their parameters. If you leave the annotation off, handle() raises a ValueError at the point of decoration, since nothing names the type a request must match. Each handle() subtracts one Ability, so half still needs an Ask and full needs nothing. Naming the two stages also gives the type checker a place to report what remains, as The type checker decides what survives handling shows.

Now compare this listing to ask_tell.py again. The by-hand version puts two objects in every signature. The Stateless version threads nothing. greet() takes no arguments, and the two Effects live in the return type where a type checker verifies them. That second channel in the signature is the one Effect Management says an EMS needs.

The whole library is visible in two_way_generator.py from Generators. An Effect is a generator, so you can drive one yourself, as hand_driven.py in Nothing Runs Yet does with that chapter’s greet("Alice"). next() on that Effect produces a Need object carrying the requested type, the way interview() yields "name". send(Console()) then answers that request and resumes the body, which prints the greeting and finishes. Every tool in the library packages those two calls. handle() is drive() with a type lookup in place of the dictionary, run() is that loop applied after the last handler, and supply() is handle() prepackaged for Need: a handler whose answer to Need[T] is whichever supplied instance is a T.

Scripting an Unpredictable Source

Every handler so far gives the same answer at every request. supply() binds one instance for the whole run, and scripted returns "Alice" no matter how many times greet() requests a name. A handler is an ordinary function, so it can answer differently at each request. Answering differently at each request makes an unpredictable source testable.

A Coin Toss

Tossing a coin is a side cause. The program reads something from outside, and the reading does not repeat. If you turn it into an Ability, the reading moves into a handler:

# coin_toss.py
import random
from stateless import Ability, Depend, handle, run

class Flip(Ability[bool]):
    pass

def flip() -> Depend[Flip, bool]:
    result: bool = yield from Flip()
    return result

def count_heads(tosses: int) -> Depend[Flip, int]:
    heads = 0
    for _ in range(tosses):
        if (yield from flip()):
            heads += 1
    return heads

script = iter((True, False, True, True, False))

def scripted(request: Flip) -> bool:
    return next(script)

def coin(request: Flip) -> bool:
    return random.random() < 0.5

print(run(handle(scripted)(count_heads)(5)))
#: 3
heads = run(handle(coin)(count_heads)(10_000))
print(4_000 < heads < 6_000)
#: True

count_heads() needs a Flip and produces an int. Its body does not call random, and no parameter carries a seed or a source. Ask and Tell each carry a payload the request delivers, whereas Flip’s whole content is its type and the bool it produces.

The parentheses in if (yield from flip()): are mandatory. A yield expression can appear on the right side of an assignment, as a statement of its own, or inside parentheses. An if condition is none of those, so if yield from flip(): is a syntax error.

Two handlers feed the same function. scripted walks an iterator over a preset sequence, so the sequence decides the five tosses before the program runs and the count is 3. coin calls random.random(), so ten thousand tosses come out near half heads. count_heads() cannot distinguish the two, because either answer arrives through the same send() channel.

The scripted handler holds state. next(script) produces a different value at each request, which one supplied instance cannot do. Every scripted test double has this shape: a queue handing out canned responses, a network stub that fails twice and then succeeds, or the clock in A Clock That Crosses Midnight.

That state has one trap, and it is silent. next(script) raises StopIteration once the sequence runs out. StopIteration is how a driver learns that an Effect has finished, so handle() reads the exhausted script as the end of the program. Every other exception a handler raises propagates out of run() unchanged, and this one never propagates. If you ask count_heads() for six tosses from this five-value script, run() produces None instead of a count, with no exception. That is the same silent None that An Effect Runs Once shows for a spent Effect. Indexing a list rather than walking an iterator turns the mistake into an IndexError that propagates out of run().

A Clock

Reading the current time is another side cause. A real clock answers with the present moment, so a test cannot set it to some critical time (midnight, or tomorrow). stamp() puts the current time into its output, and batch_due() decides whether a day has passed since the last run. Against a real clock neither is testable. stamp() produces a different string every minute, and batch_due() needs you to wait a day to watch it return True. Both sit in one file with the Ability and its accessor, because two more listings and a test ask the same clock different questions:

# timekeeping.py
from datetime import datetime, timedelta
from stateless import Ability, Depend

class Now(Ability[datetime]):
    pass

def now() -> Depend[Now, datetime]:
    moment: datetime = yield from Now()
    return moment

def stamp(message: str) -> Depend[Now, str]:
    moment = yield from now()
    return f"[{moment:%Y-%m-%d %H:%M}] {message}"

def batch_due(last_run: datetime) -> Depend[Now, bool]:
    moment = yield from now()
    return moment - last_run >= timedelta(hours=24)

Like Flip, Now carries no data. Its answer type is its whole content. A handler for Now returns a datetime, and now() is the accessor that declares that type. The handlers decide which moment stamp() and batch_due() receive:

# frozen_clock.py
from datetime import datetime, timedelta
from typing import Final
from stateless import handle, run
from timekeeping import Now, batch_due, stamp

LAUNCH: Final[datetime] = datetime(2026, 1, 1, 3, 0)

def frozen(request: Now) -> datetime:
    return LAUNCH

def tomorrow(request: Now) -> datetime:
    return LAUNCH + timedelta(hours=24)

print(run(handle(frozen)(stamp)("started")))
#: [2026-01-01 03:00] started
print(run(handle(frozen)(batch_due)(LAUNCH)))
#: False
print(run(handle(tomorrow)(batch_due)(LAUNCH)))
#: True

frozen() reports a single moment over and over, so stamp() produces a known string a test can compare. tomorrow reports a moment a day later, and batch_due() returns True at once. The schedule logic runs against whatever moment the handler names, in microseconds. batch_due() holds no datetime.now() call, so a test names the moment outright, and a production handler that returns datetime.now() runs the same function.

Those three runs are claims about testability, so each one becomes a test. The handler moves inside each test, and every test names the moment it needs:

# test_timekeeping.py
from collections.abc import Callable
from datetime import datetime, timedelta
from typing import Final
import pytest
from stateless import handle, run
from timekeeping import Now, batch_due, stamp

MOMENT: Final[datetime] = datetime(2026, 3, 14, 9, 30)

def at(moment: datetime) -> Callable[[Now], datetime]:
    def fixed(request: Now) -> datetime:
        return moment
    return fixed

def test_stamp_names_the_handlers_moment() -> None:
    stamped = run(handle(at(MOMENT))(stamp)("started"))
    assert stamped == "[2026-03-14 09:30] started"

@pytest.mark.parametrize("elapsed, due", [
    (timedelta(hours=23, minutes=59), False),
    (timedelta(hours=24), True),
])
def test_batch_due_at_a_full_day(elapsed: timedelta,
                                 due: bool) -> None:
    moment = MOMENT + elapsed
    is_due = run(handle(at(moment))(batch_due)(MOMENT))
    assert is_due is due

at() builds a handler from a moment, so each test freezes its own clock in one line. A real clock cannot produce the parametrized reading one minute short of a day on demand. Against a real clock you must start the test at the right minute, but here the margin is a timedelta. No fixture patches datetime, nothing sleeps, and each assertion compares values the test chose.

A Clock That Crosses Midnight

Skipping the wait is the obvious benefit of a handled clock. A handler can also produce moments that are hard to get from a real clock. Here, archive() reads the clock twice, once to name a file and once to stamp what goes in it:

# midnight.py
from datetime import datetime, timedelta
from typing import Final
from stateless import Depend, handle, run
from timekeeping import Now, now

def archive(entry: str) -> Depend[Now, tuple[str, str]]:
    opened = yield from now()
    path = f"log-{opened:%Y-%m-%d}.txt"
    stamped = yield from now()
    return path, f"[{stamped:%Y-%m-%d}] {entry}"

MIDDAY: Final[datetime] = datetime(2026, 1, 1, 12, 0)
LATE: Final[datetime] = datetime(2026, 1, 1, 23, 59, 59)
ticks = iter([LATE, LATE + timedelta(seconds=2)])

def steady(request: Now) -> datetime:
    return MIDDAY

def crossing(request: Now) -> datetime:
    return next(ticks)

print(run(handle(steady)(archive)("backup ok")))
#: ('log-2026-01-01.txt', '[2026-01-01] backup ok')
print(run(handle(crossing)(archive)("backup ok")))
#: ('log-2026-01-01.txt', '[2026-01-02] backup ok')

Under steady the two dates agree and the function looks correct. crossing answers the first request one second before midnight and the second request one second after midnight. The file then carries January 1’s name, and the entry inside it carries January 2’s date. A day of entries can end up in the wrong file, and the window for the mistake is only as wide as the gap between the two reads.

Using a real clock, you wait for that window and probably miss it. A test that runs at nine in the morning reads the clock twice on the same date and passes, so the only evidence is a bug report saying the log file is occasionally short by a few lines. With the Ability, archive() does not read a clock. It asks for a moment, and a handler chooses which moment that is. Both handlers answer the same two requests. They differ in whether midnight falls between the two requests.

crossing follows the same pattern as scripted in coin_toss.py. It walks a scripted list, so it holds state between requests, which is how it answers the same question two ways. A supplied instance answers every request the same way, and so do frozen and tomorrow, each reporting one moment however often you ask.

A Source Named in the Type

Compare these handlers to student_pairs.py in Functional Toolkits, which makes randomness repeatable a different way, by taking a seed parameter. A seed parameter works, but every function between the caller and the random.Random call must declare the parameter and pass it along. Here the return type names the source instead, and no signature between handle() and the request mentions it.

Flip and Now are both side causes, in the vocabulary of Effect Management: the function reads something from outside. The Recorder of Swapping the Implementation stands in for a side effect, where the function writes something outward. The technique is the same for both. Name each contact with the outside as an Ability and bind it at the edge to whatever the context needs. What an EMS adds is that the type checker enforces the declaration.

Switching Implementations Mid-Run

Every handler so far answers with data: a name, a bool, a datetime. A handler can also answer with an implementation, an object whose methods the program then calls, and it can choose a different one at each request. supply() cannot do that, because it binds one instance for the whole run. When the implementation a program depends on must change while the program is running, the choosing belongs in a handler.

A building’s power supply works this way. Solar carries the load while the sun is up. The battery carries it while the charge stays above a threshold. The utility grid carries it when neither can, and a diesel backup carries it when the grid is down too. Each stops for its own reason, and when one stops the building must obtain another.

Sources are ordinary objects. Each reports whether it can supply a given hour, and drawing from a source depletes it:

# power.py
from dataclasses import dataclass
from typing import Protocol
from record import record
from stateless import Ability, Depend, throws

class Drained(Exception):
    pass

class Source(Protocol):
    def available(self, hour: int) -> bool: ...
    def deplete(self) -> None: ...

class Solar:
    def available(self, hour: int) -> bool:
        return 6 <= hour < 19
    def deplete(self) -> None:
        pass

@dataclass
class Battery:
    charge: int
    def available(self, hour: int) -> bool:
        return self.charge >= 20
    def deplete(self) -> None:
        self.charge -= 20

@dataclass
class Grid:
    outage: range
    def available(self, hour: int) -> bool:
        return hour not in self.outage
    def deplete(self) -> None:
        pass

@dataclass
class Backup:
    fuel: int
    def available(self, hour: int) -> bool:
        return self.fuel > 0
    def deplete(self) -> None:
        self.fuel -= 1

@record(slots=False)
class Outlet(Ability[Source]):
    hour: int

def plug(hour: int) -> Depend[Outlet, Source]:
    source: Source = yield from Outlet(hour)
    return source

@throws(Drained)
def draw(source: Source, hour: int) -> None:
    if not source.available(hour):
        raise Drained(type(source).__name__)
    source.deplete()

Outlet is an Ability whose handler returns a Source, and the request carries the hour so the handler can check which sources are available at that hour. Ask carries a prompt for the same reason.

Source carries no @runtime_checkable, because nothing calls isinstance() against it. That decorator matters where supply() matches an instance to a requested class. Here handle() matches on the Ability’s own type, Outlet, and the Source that comes back is only a return value.

draw() is the lifting wrapper. It asks the source whether it can still supply, and @throws lifts the refusal into the error channel.

Here is the consumer and the handler that feeds it:

# microgrid.py
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from power import Drained, Outlet, Source, draw, plug
from stateless import Depend, catch

class Blackout(Exception):
    pass

def controller(
    order: tuple[Source, ...],
) -> Callable[[Outlet], Source]:
    def choose(request: Outlet) -> Source:
        for source in order:
            if source.available(request.hour):
                return source
        raise Blackout(request.hour)
    return choose

@contextmanager
def connected(source: Source) -> Iterator[Source]:
    name = type(source).__name__
    print(f"{name} online")
    try:
        yield source
    finally:
        print(f"{name} offline")

def run_load(start: int,
             hours: int) -> Depend[Outlet, None]:
    caught = catch(Drained)
    hour, remaining = start, hours
    while remaining:
        source = yield from plug(hour)
        with connected(source) as power:
            while remaining:
                failure = yield from caught(draw)(
                    power, hour)
                if failure is not None:
                    break
                print(f"  {hour}:00")
                hour += 1
                remaining -= 1

Read run_load() from the outside in. The outer loop obtains a source, the inner one draws from it hour after hour, and connected() brackets the pair. draw() returns None on success, so anything else in failure is the Drained that ends the inner loop. The outer loop then calls plug() again for a replacement. The hour stays put on that path, so the replacement supplies the hour the failed source refused.

connected() is an ordinary context manager, and its with block sits inside the generator body. Acquiring and releasing within one Effect works, since the block opens and closes between two yield from expressions in the same function. Stateless provides no way to acquire a resource in one Effect and release it after a later Effect finishes. Running Effects in Parallel revisits that limit.

One thing stays outside the types. choose() raises a Blackout when no source can supply the hour, and that is ordinary code raising an ordinary exception. No @throws lifts it, so it propagates out of run() and no signature mentions it. catch() matches values an Effect yields, and a handler runs in the driver, outside the Effect, so a catch() around this program lets the Blackout through.

The Blackout also leaves run_load() suspended at its plug() request, since the exception propagates out through handle() and bypasses the generator. In this program the with block has closed by then, because plug() sits outside it. A handler that raises an exception while a with block is open leaves that block’s exit waiting until Python reclaims the abandoned generator, which can come after the except clause that handled the exception has finished.

# microgrid_demo.py
from microgrid import controller, run_load
from power import Backup, Battery, Grid, Solar
from stateless import handle, run

def site() -> tuple[Solar, Battery, Grid, Backup]:
    return (Solar(), Battery(40), Grid(range(22, 24)),
            Backup(3))

solar, battery, grid, backup = site()
sun_first = controller((solar, battery, grid, backup))
run(handle(sun_first)(run_load)(17, 6))
#: Solar online
#:   17:00
#:   18:00
#: Solar offline
#: Battery online
#:   19:00
#:   20:00
#: Battery offline
#: Grid online
#:   21:00
#: Grid offline
#: Backup online
#:   22:00
#: Backup offline
solar, battery, grid, backup = site()
battery_first = controller((battery, solar, grid, backup))
run(handle(battery_first)(run_load)(17, 4))
#: Battery online
#:   17:00
#:   18:00
#: Battery offline
#: Grid online
#:   19:00
#:   20:00
#: Grid offline

The first trace is an evening. Solar covers two hours and stops at sunset. The battery covers two more and stops when its charge falls below the threshold. The grid covers one and stops when the outage begins at 22:00. The backup covers the last.

run_load() names none of the four sources, and that is why the second run prints a different trace from the same code. battery_first puts the battery ahead of the sun, so the charge drains first and the grid supplies the load from 19:00. Besides the shorter run, the order of the tuple passed to controller() is the one difference between the two runs. Priority lives in controller(), thresholds and the outage schedule in the sources themselves, and run_load() decides when to stop drawing from the source it holds.

The load’s declared dependency stays the same. Depend[Outlet, None] says it needs an Outlet from the first hour to the last, and that type appears once. What changes is the object answering the need: four objects in turn during the first run. A dependency bound before the program starts gives one answer for the whole run, and here the right answer changes with the hour. In microgrid.py the binding is a call to choose(), which checks each source’s available() at each request. Swapping the Implementation swaps an implementation between runs. microgrid.py swaps one during a run. The consumer names no source, so nothing in it changes.

State as an Ability

Each Ability so far moves information in one direction. Flip and Now read from outside, so they are side causes. Tell writes outward, so it is a side effect. Shared mutable state is both at once, because whoever holds it must read it and write it back. An Ability declares one answer type, so a cell of state becomes a pair of Abilities. Reading answers with the stored value, and writing carries a new value and answers with nothing.

# wallet.py
from collections.abc import Callable
from dataclasses import dataclass
from record import record
from stateless import Ability, Depend

class Get(Ability[int]):
    pass

@record(slots=False)
class Put(Ability[None]):
    amount: int

def get() -> Depend[Get, int]:
    amount: int = yield from Get()
    return amount

def put(amount: int) -> Depend[Put, None]:
    yield from Put(amount)

def purchase(price: int) -> Depend[Get | Put, bool]:
    funds = yield from get()
    if funds < price:
        return False
    yield from put(funds - price)
    return True

def spree(prices: tuple[int, ...]) -> Depend[
    Get | Put, int
]:
    bought = 0
    for price in prices:
        if (yield from purchase(price)):
            bought += 1
    return bought

@dataclass
class Cell:
    amount: int

def ledger(cell: Cell) -> tuple[
    Callable[[Get], int], Callable[[Put], None]
]:
    def read(request: Get) -> int:
        return cell.amount
    def write(request: Put) -> None:
        cell.amount = request.amount
    return read, write

Get has Flip’s shape: no payload, and the answer type is its whole content. Put has Tell’s: payload out, nothing back. purchase() is the function that uses both. It reads, decides, and writes, and the decision sits between the two requests, in code that mentions no cell.

purchase()’s signature declares the shared state. Depend[Get | Put, bool] tells a caller this function reads or writes something that outlives it. spree() composes purchases, and its signature carries the same union.

The handlers own the cell. ledger() builds read() and write() from one Cell, the way at() builds a clock from a moment.

A test calls ledger() on a Cell of its own, chains the pair through the named stages of Abilities Are Not Special, and asserts on the count spree() returns and on what that cell holds afterward:

# test_wallet.py
from stateless import handle, run
from wallet import Cell, ledger, spree

def test_spree_spends_from_the_tests_cell() -> None:
    cell = Cell(100)
    read, write = ledger(cell)
    half = handle(read)(spree)
    shop = handle(write)(half)
    assert run(shop((60, 50, 30, 20))) == 2
    assert cell.amount == 10

After the run, the cell holds what the program left. Two purchases went through, and 10 remained. The test asserts on the cell it built, so nothing needs resetting between tests and two tests can run in either order.

When one function owns a number, a local variable is the right tool, and count_heads() keeps its count in one. The pair of Abilities is for separate functions that share the cell, as purchase() and any other spender do, without a parameter threaded through every signature between them.

This pattern has a name. Treatments of algebraic effects open with the State effect, get and put as its two operations. Stateless’s machinery is enough to build that effect.

One warning comes with the pattern. Nothing guards the cell. Forking two Effects that share a Cell interleaves their reads and writes, and no type reports the race. ZIO’s Ref is this cell with atomic update built in. Stateless has no equivalent, so under fork() the cell is as exposed as any Python global.

Composing a Program

This small application fetches a headline, finds a topic worth researching within that headline, and looks up that topic. Each step needs something or can fail, and no step names an implementation:

# research.py
from typing import Final, Protocol, runtime_checkable
from stateless import Effect, Need, need, throws

class Unavailable(Exception):
    pass

class NotInteresting(Exception):
    pass

class NoArticle(Exception):
    pass

@runtime_checkable
class Feed(Protocol):
    def latest(self) -> str: ...

@runtime_checkable
class Encyclopedia(Protocol):
    def article(self, topic: str) -> str: ...

TOPICS: Final[tuple[str, ...]] = ("stock market", "genome")

@throws(Unavailable)
def fetch(feed: Feed) -> str:
    return feed.latest()

@throws(NotInteresting)
def topic_of(headline: str) -> str:
    for candidate in TOPICS:
        if candidate in headline:
            return candidate
    raise NotInteresting(headline)

@throws(NoArticle)
def look_up(book: Encyclopedia, topic: str) -> str:
    return book.article(topic)

def research() -> Effect[
    Need[Feed] | Need[Encyclopedia],
    Unavailable | NotInteresting | NoArticle,
    str,
]:
    feed = yield from need(Feed)
    headline = yield from fetch(feed)
    topic = yield from topic_of(headline)
    book = yield from need(Encyclopedia)
    article = yield from look_up(book, topic)
    return article

The research() signature declares what the program needs and how it can fail. research() reads a Feed and an Encyclopedia from outside and can fail three ways. The three @throws functions are the pattern for bringing ordinary code in. fetch() and look_up() call ordinary methods rather than Effects, and the decorator lifts what they raise into the channel. topic_of() reads nothing from outside and writes nothing, so it declares no Ability.

research()’s signature is also the only place its failures appear, and the body requests each dependency by its type. Nothing in the body mentions a network, a file, or a print, and research() performs no work when called.

fetch() and look_up() take their dependencies as parameters, so they are ordinary functions rather than generator functions. Taking dependencies as parameters is a choice, not a requirement. @throws also decorates a generator function, so the request and the failure can live in one function:

# fetch_effectful.py
from research import Feed, Unavailable
from stateless import Depend, Need, need, throws

@throws(Unavailable)
def fetch_headline() -> Depend[Need[Feed], str]:
    feed = yield from need(Feed)
    return feed.latest()

The decorator adds the error the same way it does for score() in The Error Channel. ty reports fetch_headline as () -> Generator[Need[Feed] | Unavailable, Any, str], which is Effect[Need[Feed], Unavailable, str].

research.py keeps the request and the failure in separate functions, because a function that receives everything it uses as arguments is easier to test on its own, and because the split keeps the Ability requests in one place. Either shape type-checks and either propagates correctly.

newswire.py defines two feeds and an encyclopedia to supply:

# newswire.py
from dataclasses import dataclass
from research import NoArticle, Unavailable

@dataclass
class Wire:
    headline: str
    def latest(self) -> str:
        print("feed: fetching")
        return self.headline

class DeadWire:
    def latest(self) -> str:
        raise Unavailable("offline")

@dataclass
class Library:
    articles: dict[str, str]
    def article(self, topic: str) -> str:
        print(f"library: looking up {topic}")
        if topic not in self.articles:
            raise NoArticle(topic)
        return self.articles[topic]

Wire and Library are structural implementations, so supply(Wire(...), Library(...)) builds handlers for Need[Wire] and Need[Library], the mismatch that Supplying an Interface fixes with as_type().

Now supply research()’s environment:

# scenarios.py
from typing import Final, assert_never
from newswire import DeadWire, Library, Wire
from research import (Encyclopedia, Feed, NoArticle,
                      NotInteresting, Unavailable, research)
from stateless import Depend, Need, catch, run, supply

def report() -> Depend[
    Need[Feed] | Need[Encyclopedia], str
]:
    caught = catch(Unavailable, NotInteresting, NoArticle)
    found: str | Unavailable | NotInteresting | NoArticle
    found = yield from caught(research)()
    match found:
        case Unavailable():
            return "no headline today"
        case NotInteresting():
            return "nothing worth researching"
        case NoArticle():
            return "no article on that topic"
        case str():
            return found
        case _:
            assert_never(found)

STOCKS: Final[Wire] = Wire("stock market rising")
WEATHER: Final[Wire] = Wire("mild and cloudy")
SHELF: Final[Library] = Library(
    {"stock market": "a history"})
EMPTY: Final[Library] = Library({})

def outcome(feed: Feed, book: Encyclopedia) -> str:
    return run(supply(feed, book)(report)())

print(outcome(STOCKS, SHELF))  # [1]
#: feed: fetching
#: library: looking up stock market
#: a history
print(outcome(WEATHER, SHELF))  # [2]
#: feed: fetching
#: nothing worth researching
print(outcome(STOCKS, EMPTY))  # [3]
#: feed: fetching
#: library: looking up stock market
#: no article on that topic
print(outcome(DeadWire(), SHELF))  # [4]
#: no headline today

Four runs of one program, differing in what you supply. The run at [1] finds its article. The run at [2] exercises NotInteresting, [3] exercises NoArticle, and [4] exercises Unavailable, so the runs cover every failure the signature declares. Effect Oriented Programming calls each such set of bindings a scenario, and here a scenario is nothing more than arguments to supply().

Every printed line in that trace comes from a supplied implementation, because the pipeline holds no output of its own, so the trace also records where each run stopped. The run at [2] stops after feed: fetching. topic_of() yields a NotInteresting, and that failure ends research() where it stands. The need(Encyclopedia) on the next line does not run, and nothing calls Library.article(). catch() receives that failure and report() matches on it as a value, which is why the run still prints a message. A failure ends the remaining steps the way a raised exception does, and no step tests for it. Where the run stops depends on where the failure arises. The run at [4] prints no trace, since DeadWire.latest() raises Unavailable without printing anything. The run at [3] reaches the library and fails there.

report() handles the two channels differently. catch() empties the error channel, so report() cannot fail. report() still declares both Abilities, because catching an error does nothing about a dependency. If you annotate report() as Success[str], the type checker names the yield from that still carries Need[Feed] | Need[Encyclopedia]. supply() empties that half, and run() accepts what remains.

Declaring outcome()’s parameters as Feed and Encyclopedia does the job of as_type() at the boundary, without a cast.

The Success Path

research() handles no errors. Its body is a straight run of six lines, each naming the next step, and no line tests whether the previous one worked. The error channel makes that straight run possible. Here is the same pipeline with the failures handled where they arise:

# research_by_hand.py
from research import (TOPICS, Encyclopedia, Feed, NoArticle,
                      NotInteresting, Unavailable)

def topic_of(headline: str) -> str:
    for candidate in TOPICS:
        if candidate in headline:
            return candidate
    raise NotInteresting(headline)

def research_and_report(
    feed: Feed, book: Encyclopedia
) -> str:
    try:
        headline = feed.latest()
    except Unavailable:
        return "no headline today"
    try:
        topic = topic_of(headline)
    except NotInteresting:
        return "nothing worth researching"
    try:
        return book.article(topic)
    except NoArticle:
        return "no article on that topic"

topic_of() appears again in full because research.py’s version carries a decorator. The decorated version returns an Effect, and ordinary try/except code that calls it gets that Effect back in place of a topic.

Three lines of work sit inside nine lines of handling. The pipeline is in there, spread across three try blocks. The Effect version moves those nine lines into report(), one match over the failures instead of a try at each step. The name research_and_report() records the merge the by-hand version forces: research() and report() in one function.

Both versions short-circuit. The by-hand one returns early. The Effect one stops at the failure, and the driver never resumes the generator. The difference is who writes the branch that short-circuits.

The comparison has limits. At this size the by-hand version is respectable, and a reader may prefer it. Two differences outlast the size argument. The by-hand signature, (Feed, Encyclopedia) -> str, mentions none of the three failures, so adding a fourth failure changes nothing in the signature. And the handling interleaves with the logic. research_and_report() decides both what to do about a failure and what to say about it. The Effect version separates those, so a second caller can catch the same three failures and choose different messages, retry the whole pipeline, or let one failure through to the edge, without editing the pipeline.

A Second Way In and Out

The error channel you have seen has one way in and one way out. @throws lifts what ordinary code raises, and catch() takes named errors back out as values. The library provides a second of each.

Failing from Inside an Effect

throw() is the counterpart of success(). Where success(value) builds a description that produces a value, throw(reason) builds one that fails. Its type is Try[E, Never], the Try alias, with Never recording that no value can come out of it. yield from sends that failure into the channel of the Effect that contains it:

# fetch_guarded.py
from dataclasses import dataclass
from research import Feed, Unavailable, fetch
from stateless import (Effect, Need, catch, need, run,
                       supply, throw)

class Empty(Exception):
    pass

def fetch_nonempty() -> Effect[
    Need[Feed], Unavailable | Empty, str
]:
    feed = yield from need(Feed)
    headline = yield from fetch(feed)
    if not headline:
        yield from throw(Empty())
    return headline

@dataclass
class Ticker:
    headline: str
    def latest(self) -> str:
        return self.headline

def edge(feed: Feed) -> str | Unavailable | Empty:
    guarded = catch(Unavailable, Empty)(fetch_nonempty)
    return run(supply(feed)(guarded)())

print(edge(Ticker("markets close mixed")))
#: markets close mixed
print(type(edge(Ticker(""))).__name__)
#: Empty

fetch_nonempty() can fail either way. Unavailable originates in ordinary code. latest() raises it, and the @throws on fetch() lifts it. Empty originates here, in the generator, where the headline is available to inspect. No decorator takes part and nothing raises an exception. throw(Empty()) yields the failure the way Ask(prompt) yields a request, and the driver receives it. Execution stops there. A driver that receives a failure stops sending, so anything after a yield from throw(...) is unreachable. The Never in throw()’s type records that.

The difference between the two ways in is what the type checker verifies. If you change Empty() to some undeclared exception, ty flags the yield from where it stands, because the yielded type falls outside the declared channel, the invalid-yield that Dependencies That Need Dependencies shows in full. A raise gets no such check. @throws lifts the types it names, and anything else propagates untracked. No diagnostic connects the decorator’s list to what the body raises. A failure that enters through throw() is in the type system from the moment it exists.

Catching the Whole Channel

catch() names what it takes. catch_all() takes whatever the channel declares:

# catch_everything.py
from dataclasses import dataclass
from research import (Encyclopedia, Feed, NoArticle,
                      NotInteresting, Unavailable, research)
from stateless import run, supply
from stateless.effect import catch_all

@dataclass
class Bulletin:
    headline: str
    def latest(self) -> str:
        return self.headline

class BareShelf:
    def article(self, topic: str) -> str:
        raise NoArticle(topic)

def outcome(
    feed: Feed, book: Encyclopedia
) -> str | Unavailable | NotInteresting | NoArticle:
    bound = supply(feed, book)(research)
    return run(catch_all(bound)())

dull = outcome(Bulletin("mild and cloudy"), BareShelf())
print(type(dull).__name__)
#: NotInteresting
missing = outcome(Bulletin("genome mapped"), BareShelf())
print(type(missing).__name__)
#: NoArticle

outcome() is the boundary function scenarios.py uses, with the same upcasting annotations. Its result type is the union report() builds by naming all three errors, and this listing names none of them at the catch. Two failures from two different sources come back as values through the one catch_all().

outcome() supplies first and catches second, the reverse of scenarios.py, and both the runtime and the type checker accept either order. A handler passes error values upward untouched, so supply()’s driver relays the failures to the catch either way. Under ty 0.0.84 both orders infer the same result type. The Ability channel is where the orders differ. supply(feed, book)(catch_all(research)) comes back with Never there, and catch_all(supply(feed, book)(research)) with Unknown: the same two calls in the other order. bound has a name so that you can reveal its type on its own, and The type checker decides what survives handling explains the difference.

catch() and catch_all() differ in what they do with a new failure. When research() gains a fourth error, report()’s named catch() leaves it in the channel, so report()’s declared type no longer matches and the type checker points at the yield from. You decide whether to catch the newcomer or declare it. catch_all() absorbs it into the result union instead, so the guard must sit downstream, in an annotation that writes the union out, as outcome()’s return type does, or in a match that ends with assert_never(). Without such a guard, the new failure becomes a value that reaches the caller unexamined. catch() makes each failure an explicit decision. catch_all() decides for all of them at once.

Two cautions. catch_all() comes from stateless.effect, since the package root does not export it. And catch_all() leaves the guarantee where it was. It matches the failures @throws lifted into the channel, and an unlifted failure propagates past it as it does past catch().

Dependencies That Need Dependencies

research() asks for two things, and both are leaves. Nothing needs building before you can supply a Feed or an Encyclopedia. Real graphs nest. Toast needs bread and a toaster, and bread needs dough and an oven.

# bakery.py
from record import record
from stateless import Depend, Need, need, run, supply

@record
class Dough:
    flour: str
    def risen(self) -> str:
        print("dough: risen")
        return f"{self.flour} dough"

@record
class Oven:
    celsius: int
    def bake(self, dough: str) -> str:
        print(f"oven: baking at {self.celsius}")
        return f"loaf of {dough}"

@record
class Toaster:
    setting: int
    def brown(self, loaf: str) -> str:
        print(f"toaster: setting {self.setting}")
        return f"toasted {loaf}"

def bread() -> Depend[Need[Dough] | Need[Oven], str]:
    dough = yield from need(Dough)
    oven = yield from need(Oven)
    return oven.bake(dough.risen())

def toast() -> Depend[
    Need[Dough] | Need[Oven] | Need[Toaster], str
]:
    loaf = yield from bread()
    toaster = yield from need(Toaster)
    return toaster.brown(loaf)

kitchen = supply(Dough("rye"), Oven(220), Toaster(3))
print(run(kitchen(toast)()))
#: dough: risen
#: oven: baking at 220
#: toaster: setting 3
#: toasted loaf of rye dough

You supply appliances and make products, and the listing keeps the two apart. Dough, Oven, and Toaster are the leaves, so supply() binds one instance of each. The loaf is not a leaf. bread() is an Effect that produces a loaf, so toast() obtains one by writing yield from bread() rather than by asking for a Need[Bread]. Nothing supplies a loaf, because no loaf exists when supply() runs.

The signature carries the graph, flattened into a union. toast() declares all three leaves although its body names one of them. yield from bread() carries Need[Dough] and Need[Oven] from the inner Effect into toast()’s channel, the way yield from relays whatever the inner generator yields. The type checker maintains that union. If you declare toast() with Need[Toaster] alone, ty points at the delegation:

error[invalid-yield]: Yield expression type does not match annotation
  --> bakery.py:34:23
   |
31 |   def toast() -> Depend[
   |  ________________-
32 | |     Need[Toaster], str
33 | | ]:
   | |_- Function annotated with yield type
   |     `Need[Toaster]` here
34 |       loaf = yield from bread()
   |                         ^^^^^^^ expression of type
   |                         `Need[Dough] | Need[Oven]`,
   |                         expected `Need[Toaster]`

The type checker checks the other end too. If you leave Oven(220) out of supply(), the checker rejects the run() call, finding a Generator[Need[Oven], Any, str] where it expected an empty Ability channel. That is the rejection Forgetting to Supply shows, now arising from a dependency two levels down.

Oven and Toaster are distinct types, so the ambiguity of When Two Implementations Match cannot arise here. Effect Oriented Programming builds this same toast in ZIO, where both appliances are a HeatSource and ZIO must report the clash.

Here is what ZIO does that Stateless cannot. In that book’s version, Bread.homeMade is a ZLayer: a constructor that is an Effect. It can print, it can fail, and you can retry it. The compiler resolves it into a tree with Oven and Dough beneath it. You provide that layer rather than a finished loaf. Stateless has no such thing. supply() matches instances that exist, and handle() answers an Ability with an ordinary function, so a constructor cannot be an Effect. That is why bakery.py has the shape it has. Leaves bind at the edge, products come from an explicit yield from, and the graph you can read is the union in the signature.

Supplying a Whole Cast

The bakery graph goes deep. Three appliances, two of them reached through bread(). The next example goes wide instead. Abstract Factories builds a gaming environment where a GameElementFactory returns a matched Character and Obstacle, and a GameEnvironment plays whatever that factory produces. Here each kind of actor is an Ability the program requests:

# quest.py
from typing import Protocol, runtime_checkable
from stateless import Depend, Need, need

@runtime_checkable
class Narrator(Protocol):
    def say(self, line: str) -> None: ...

@runtime_checkable
class Hero(Protocol):
    def name(self) -> str: ...
    def approach(self, obstacle: str) -> str: ...

@runtime_checkable
class Obstacle(Protocol):
    def blocks(self) -> str: ...

def encounter() -> Depend[
    Need[Narrator] | Need[Hero] | Need[Obstacle], None
]:
    narrator = yield from need(Narrator)
    hero = yield from need(Hero)
    obstacle = yield from need(Obstacle)
    narrator.say(f"{hero.name()} arrives")
    narrator.say(hero.approach(obstacle.blocks()))

encounter() holds all the engine’s logic. It constructs no GameEnvironment, holds no factory, and the only types it mentions are the three Protocols. It prints nothing itself. Output is an Ability like Hero and Obstacle. Narrator is one of the three requests, so the code that supplies it chooses whether a line prints, goes into a list, or disappears. As Retrofitting an Effect recommends, the union appears in full rather than as an alias.

Each Ability needs a shape of its own. You could rename Obstacle.blocks() to name(), which Hero already declares. Hero stays distinct even then, since it also declares approach(), but every hero then satisfies Obstacle as well, and play() supplies the hero first, so the hero answers the request for an obstacle. Two Protocols with matching methods leave argument order to decide which request each one answers, the ambiguity of When Two Implementations Match.

The cast is a set of ordinary classes that inherit nothing:

# casts.py
from quest import Hero, Narrator, Obstacle, encounter
from stateless import run, supply

class Kitty:
    def name(self) -> str: return "Kitty"
    def approach(self, obstacle: str) -> str:
        return f"and bats at the {obstacle}"

class Puzzle:
    def blocks(self) -> str: return "puzzle"

class Warrior:
    def name(self) -> str: return "Warrior"
    def approach(self, obstacle: str) -> str:
        return f"and battles the {obstacle}"

class Weapon:
    def blocks(self) -> str: return "nasty weapon"

def play(
    narrator: Narrator, hero: Hero, obstacle: Obstacle
) -> None:
    cast = supply(narrator, hero, obstacle)
    run(cast(encounter)())

def kitties_and_puzzles(narrator: Narrator) -> None:
    play(narrator, Kitty(), Puzzle())

def warriors_and_weapons(narrator: Narrator) -> None:
    play(narrator, Warrior(), Weapon())

play() is the boundary function of Composing a Program, grown from two parameters to three. Its annotations do the upcasting, so each actor reaches supply() as its Protocol type, the job as_type() does elsewhere. Its body is the one place in the program that binds an implementation to an Ability.

kitties_and_puzzles() and warriors_and_weapons() are what the two concrete factories became. Each was a class with a method per product. Each is now a function that hands play() a matched set. No parallel hierarchy remains. Kitty does not extend a Character base class, Puzzle does not extend an Obstacle base class, and the engine names no class:

# two_games.py
from dataclasses import dataclass, field
from casts import (Kitty, Weapon, kitties_and_puzzles, play,
                   warriors_and_weapons)

class Loud:
    def say(self, line: str) -> None: print(line)

@dataclass
class Script:
    lines: list[str] = field(default_factory=list)
    def say(self, line: str) -> None:
        self.lines.append(line)

kitties_and_puzzles(Loud())
#: Kitty arrives
#: and bats at the puzzle
warriors_and_weapons(Loud())
#: Warrior arrives
#: and battles the nasty weapon
play(Loud(), Kitty(), Weapon())
#: Kitty arrives
#: and bats at the nasty weapon
script = Script()
kitties_and_puzzles(script)
print(len(script.lines), script.lines[1])
#: 2 and bats at the puzzle

The fourth run swaps one cast member and captures the output. Script records what arrives, so a test reads the lines back as a list with no capsys and no monkeypatching. The engine holds no printing to intercept. test_greeter.py in Swapping the Implementation makes the same swap with one Ability rather than three.

The Unmatched Cast

The third run mixes the casts, and nothing reports it. A Kitty bats at a Weapon. The unmatched cast type-checks, and it runs. That is a real loss against the Abstract Factory, whose purpose is families of matched products. KittiesAndPuzzles.make_obstacle() cannot return a Weapon, because the pairing lives inside the class. supply() takes a flat list and checks each argument against one Ability, not against the others. The matched set comes back only if you write it down, and kitties_and_puzzles() does. The guarantee moves from a class hierarchy into a two-line function. Know which of those you are getting.

The Nine-Argument Ceiling

One supply() call can type-check at most nine actors. supply()’s declaration carries overloads for one through nine values, so a tenth argument matches none of them:

error[no-matching-overload]: No overload of function `supply`
matches arguments

The call still runs correctly, since the implementation is variadic, but the type checker no longer verifies the call. Two chained handlers keep the checking. supply() some of the actors, apply that handler to the Effect, then supply() the rest to what remains, the layered supply of Layering Handlers. An Effect that asks for ten separate things is usually two Effects.

Adding Behavior to an Existing Effect

The Success Path says that a caller can retry a pipeline without editing it. Stateless provides a few decorators that add such behavior. Retry is the one to study, because of what it does to the type.

retry() and a Flaky Database

Database fails a set number of times before working, so the example is repeatable:

# flaky.py
from dataclasses import dataclass
from stateless import Effect, Need, need, throws

class Crashed(Exception):
    pass

@dataclass
class Database:
    failures: int
    attempts: int = 0
    def save(self, user: str) -> str:
        self.attempts += 1
        print(f"attempt {self.attempts}: saving {user}")
        if self.attempts <= self.failures:
            raise Crashed("database crashed")
        return f"{user} saved"

@throws(Crashed)
def store(db: Database, user: str) -> str:
    return db.save(user)

def save_user(user: str) -> Effect[
    Need[Database], Crashed, str
]:
    db = yield from need(Database)
    result = yield from store(db, user)
    return result

retry() takes a Schedule and decorates a function that returns an Effect:

# retrying.py
from datetime import timedelta
from flaky import Crashed, Database, save_user
from stateless import catch, retry, run, supply
from stateless.functions import RetryError
from stateless.schedule import recurs, spaced
from stateless.time import Time

once = catch(Crashed)(save_user)
print(run(supply(Database(failures=2))(once)("Morty")))
#: attempt 1: saving Morty
#: database crashed
three = recurs(3, spaced(timedelta(milliseconds=1)))
retried = retry(three)(save_user)
print(run(
    supply(Database(failures=2), Time())(retried)("Morty")))
#: attempt 1: saving Morty
#: attempt 2: saving Morty
#: attempt 3: saving Morty
#: Morty saved
caught = catch(RetryError)(retried)
outcome = run(
    supply(Database(failures=9), Time())(caught)("Morty"))
#: attempt 1: saving Morty
#: attempt 2: saving Morty
#: attempt 3: saving Morty
print(type(outcome).__name__)
#: RetryError

The first run is the baseline: one attempt, no retry, and it fails. three comes from the only two schedule combinators the library has. spaced() yields the same interval forever, and recurs() stops it after n yields. Three attempts against a database that fails twice succeed on the third, and three attempts against one that always fails produce a RetryError holding every failure. save_user() stays unchanged through all of it.

retry() decorates the function save_user, not the Effect save_user("Morty"), for the reason An Effect Runs Once gives. The Effect is a generator, one run() spends it, and only the function can build a second description.

What Retry Cannot Judge

Read the trace before you use retry() on real code. Each attempt line is Database.save() running again, so the decorated function runs its whole body once per attempt. Retrying a charge or an append duplicates it. Nothing in the type says whether a retry is safe, because Effect[A, E, R] tracks what a function needs and how it fails, not whether running it twice means the same as running it once. That judgment stays with you.

retry() is also all-or-nothing over the error channel. Its body catches every declared error and retries on any of them, with no way to name the ones worth another attempt. Retrying research() with the WEATHER feed of Composing a Program shows the problem. A WEATHER headline names no known topic, so topic_of() raises NotInteresting on every attempt, deterministically. retry() treats that failure like one worth retrying. Three attempts fetch the headline three times and collect three identical NotInteresting errors in the RetryError. Retry what can change between attempts, and let catch() take a deterministic failure out before it reaches retry().

What Retry Costs the Signature

save_user() is (str) -> Effect[Need[Database], Crashed, str]. Under reveal_type(), retried is:

(user: str) -> Generator[
    Need[Database] | Need[Time] | Async | RetryError[Crashed],
    Any,
    str,
]

Three changes, none of them silent. The error becomes RetryError[Crashed], which is why the third run catches RetryError rather than Crashed. Async appears because waiting between attempts is asynchronous. And Need[Time] appears, which is why supply() gains a Time(). Retrying needs a clock, and the signature says so. If you leave the Time() out, the type checker rejects the run() call. That Need[Time] is the thesis of both chapters applied to a cross-cutting concern. In a system with untracked Effects, adding retry to a hundred call sites changes no signature. Here adding retry changes a type, and every caller that runs the retried function without a Time fails the check.

The renamed error allows a mistake the type checker accepts. If you write catch(Crashed)(retried), catching the original error, the checker reports nothing. The result type gains a Crashed branch that cannot occur, RetryError[Crashed] stays in the error channel, and at runtime the failure propagates past the useless catch() and out of run(). catch() must name what the channel holds at the point of decoration, and after retry() that is RetryError[Crashed], not Crashed.

One library omission: RetryError declares an errors attribute, and retry() builds one as RetryError(tuple(errors)). The collected failures therefore live in outcome.args[0], and outcome.errors raises AttributeError.

repeat() and memoize()

repeat() is the sibling that runs an Effect on a schedule and collects every result:

# repeating.py
from datetime import timedelta
from flaky import Database, save_user
from stateless import repeat, run, supply
from stateless.schedule import recurs, spaced
from stateless.time import Time

three = recurs(3, spaced(timedelta(milliseconds=1)))
repeated = repeat(three)(save_user)
env = supply(Database(failures=0), Time())
print(run(env(repeated)("Morty")))
#: attempt 1: saving Morty
#: attempt 2: saving Morty
#: attempt 3: saving Morty
#: ('Morty saved', 'Morty saved', 'Morty saved')

The same schedule that governed three attempts now governs three runs, and the tuple holds every result in order. Where retry() stops at the first success, repeat() continues until the schedule runs out. A failure on any run propagates unchanged rather than becoming a RetryError.

memoize() solves the spent-generator problem:

# memoizing.py
from flaky import Database, save_user
from stateless import memoize, run, supply

db = Database(failures=0)
bound = supply(db)(memoize(save_user))
print(run(bound("Morty")))
#: attempt 1: saving Morty
#: Morty saved
print(run(bound("Morty")))
#: Morty saved
print(f"attempts: {db.attempts}")
#: attempts: 1

memoize() caches by argument the way functools.lru_cache does. It wraps the Effect in an object that records the result and replays it rather than driving the spent generator again. That wrapper exists because a generator cannot run twice, which is the same fact that makes retry() decorate the function.

That cache key is the argument, not the environment. memoize() wraps save_user() before supply() runs, so a different Database supplied to the same memoized call changes nothing in the cache:

# memoize_environment.py
from flaky import Database, save_user
from stateless import memoize, run, supply

db1 = Database(failures=0)
db2 = Database(failures=0)
m = memoize(save_user)
print(run(supply(db1)(m)("Morty")))
#: attempt 1: saving Morty
#: Morty saved
print(run(supply(db2)(m)("Morty")))
#: Morty saved
print(db1.attempts, db2.attempts)
#: 1 0

db2 never runs. The second call’s answer comes from db1’s cache entry, because memoize() keys on "Morty" alone. In a chapter built on swapping the environment, the caution matters. Memoize a function only where one environment serves the memoized call for its whole lifetime, the way State as an Ability’s Cell does once bound, or key the cache on the environment too.

Running Effects in Parallel

fork() submits an Effect to an Executor and returns a Task, and wait() collects the result. This is the same wait() that awaits a coroutine in Waiting on a Coroutine. It accepts a Task as readily as an awaitable:

# parallel.py
import time
from concurrent.futures import Executor, ThreadPoolExecutor
from benchmark import report
from stateless import (Async, Depend, Need, Success, Task,
                       as_type, fork, run, success, supply,
                       wait)

@fork
def slow_square(n: int) -> Success[int]:
    time.sleep(0.05)
    return success(n * n)

def squares(
    count: int,
) -> Depend[Need[Executor] | Async, list[int]]:
    tasks: list[Task[int]] = []
    for n in range(count):
        task = yield from slow_square(n)
        tasks.append(task)
    results: list[int] = []
    for task in tasks:
        value = yield from wait(task)
        results.append(value)
    return results

with ThreadPoolExecutor(max_workers=5) as pool:
    start = time.perf_counter()
    out = run(supply(as_type(Executor)(pool))(squares)(5))
    elapsed = time.perf_counter() - start
report(elapsed=elapsed)
print(out)
#: [0, 1, 4, 9, 16]
print(f"five 50ms tasks under 150ms: {elapsed < 0.15}")
#: five 50ms tasks under 150ms: True

Five tasks that each sleep 50 milliseconds finish in about the time of one. That comes from the two loops. squares() forks every task before it waits for any of them, so the five sleeps overlap. If you fork and wait inside a single loop, each wait() blocks on the task the same iteration just created, so the sleeps run one after another and take about five times as long.

The pool is an Ability, not a global, so squares() declares Need[Executor] and names no pool. Supplying a ProcessPoolExecutor moves the same work into processes, and squares() stays as written. as_type(Executor) appears for the reason it always does. ThreadPoolExecutor is the more specific type, and squares() asks for the general one.

The type checker enforces one restriction. A forked Effect must have nothing left to supply. fork()’s four overloads accept an Effect whose Ability channel holds Never, an exception type, or Async, because fork() runs the Effect with run() inside the worker. If you decorate a function that still declares a Need, the checker rejects it, listing the overloads it failed to match. Supply first, then fork.

That restriction is the only one the checker enforces. The restriction says nothing about a declared error. Two of those same four overloads accept an Effect that declares one, and both return a Task with no error type on it.

Notice who manages the pool’s lifetime. The with block sits outside run(), at the edge, in ordinary Python. Stateless has no scoping mechanism of its own, so either a resource lives in a with block outside the Effect, as the pool does here, or the supplied object owns it. The library’s own Files class opens and closes a file inside a single read_file() call. The library gives you no way to acquire a resource in one Effect and release it after a later one finishes, the flat resource management a native Effect system provides. Python’s own answer to that is ExitStack in Combining Context Managers, which holds a set of managers decided at runtime and unwinds them together. It flattens the nesting in ordinary Python, so the gap is smaller than it first appears.

The Toolkit

Here is every tool from both chapters that acts on a description. Each one builds a description, rewrites a description’s type, or executes one. Three sit outside the tables. as_type() relabels a value for the type checker and does nothing at runtime. spaced() and recurs() build the Schedule that retry() and repeat() consume. The accessors for the library’s own dependencies (print_line(), read_line(), and read_file()) each call need() for one class, so need()’s row covers them.

Five build a description:

The rest decorate a function that returns an Effect, rewriting the type that function declares:

Three rows carry a caveat. fork needs a function whose Effect has nothing left to supply, so supply first, then fork. @throws is an entry point rather than a transformation. It decorates an ordinary function that raises exceptions, turning it into one that returns an Effect. throw() instead builds the failure as a description. And catch_all comes from stateless.effect, since the package root does not export it.

Two execute an Effect whose yield channel has narrowed to Async and errors, raising a leftover error rather than returning it:

These two are the only functions that perform work, which is the description/execution split in table form.

run() Builds a Loop per Call

The rule about where to call run() and run_async() has a reason. run() is asyncio.run(run_async(effect)), building a fresh event loop for the call and tearing it down after. If you call it from inside a running loop (the shape of any async web handler), asyncio.run() raises a RuntimeError before your Effect runs:

# run_cost.py
import asyncio
import time
from benchmark import report
from stateless import Success, run, run_async, success

def bound(n: int) -> Success[int]:
    return success(n)

async def inside_loop() -> str:
    try:
        run(bound(1))
    except RuntimeError as e:
        return str(e)
    return "no error"

print(asyncio.run(inside_loop()))
#: asyncio.run() cannot be called from a running event loop

ROUNDS = 2000

def time_run() -> float:
    start = time.perf_counter()
    for _ in range(ROUNDS):
        run(bound(1))
    return (time.perf_counter() - start) / ROUNDS

async def time_run_async() -> float:
    start = time.perf_counter()
    for _ in range(ROUNDS):
        await run_async(bound(1))
    return (time.perf_counter() - start) / ROUNDS

per_run = time_run()
per_run_async = asyncio.run(time_run_async())
report(per_run=per_run, per_run_async=per_run_async)
print(f"run() at least 20x slower: "
      f"{per_run > per_run_async * 20}")
#: run() at least 20x slower: True

run_async() reuses the running loop and costs little beyond the Effect. run() builds and tears down a loop on every call, at least twenty times the cost of run_async(), by the listing’s own measure. Running the listing with --numbers prints both per-call costs. Synchronous code has no loop to reuse, so it pays that cost on every run(). From inside a running loop, run_async() is both the one that works and the one that is fast.

Where the Guarantee Stops

The guarantee has six limits.

1. Nothing stops an undeclared Effect

Success[int] claims purity, and this function breaks that claim:

# leaky_effect.py
from stateless import Success, run, success

def double(n: int) -> Success[int]:
    print(f"doubling {n}")
    return success(n * 2)

print(run(double(21)))
#: doubling 21
#: 42

That listing type-checks. Stateless verifies that declared Effects propagate. It cannot verify that you declared everything effectful, because Python’s print(), open(), and requests.get() are ordinary calls with ordinary types. A native EMS computes a function’s Effects from its body. A library’s view stops at the Effects you wrote down. The guarantee is about consistency, not completeness.

The same limit applies on the error side. Here Success[float] claims that ratio() cannot fail:

# undeclared_failure.py
from exceptions import expect
from stateless import Success, catch, run, success

def ratio(a: int, b: int) -> Success[float]:
    return success(a / b)

def caller() -> Success[float | ZeroDivisionError]:
    out: float | ZeroDivisionError
    out = yield from catch(ZeroDivisionError)(ratio)(1, 0)
    return out

expect(ZeroDivisionError, run, caller())
#: [ZeroDivisionError] division by zero

ratio() carries no @throws, so the ZeroDivisionError propagates as an ordinary raised exception, untracked. catch() matches the values an Effect yields, not exceptions the body raises, so a failure that no @throws lifts goes past catch() untouched. Watch for a failure going past catch(), because catch(ZeroDivisionError) type-checks and then does nothing. The protection appears to be there. The channel carries only what you put into it.

ZIO does not catch an undeclared failure at compile time either. Its own documentation uses the same example. def divide(a: Int, b: Int): ZIO[Any, Nothing, Int] = ZIO.succeed(a / b) declares an error type of Nothing, and a zero denominator still throws. The difference is at runtime. An exception thrown inside a computation ZIO runs becomes a defect, recorded on the Cause beside the typed error channel. sandbox() can recover a defect, and the runtime logs it as a dying fiber. Stateless has no defect channel, so the exception leaves run() as an ordinary Python exception.

2. The type checker decides what survives handling

How much of a type survives handling depends on your type checker rather than on the library. nested and full below come from the same two calls in the same order, one form nesting them and the other naming the intermediate:

# nested_handle.py
from typing import reveal_type
from ask_tell_stateless import capture, greet, scripted
from stateless import handle

nested = handle(scripted)(handle(capture)(greet))
half = handle(capture)(greet)
full = handle(scripted)(half)

if __name__ == "__main__":
    reveal_type(nested)
    reveal_type(half)
    reveal_type(full)

Under ty 0.0.84, ty check nested_handle.py reads the two forms the same way:

info[revealed-type]: Revealed type
  --> nested_handle.py:11:17
   |
11 |     reveal_type(nested)
   |                 ^^^^^^ `() -> Generator[Never, Any, None]`

info[revealed-type]: Revealed type
  --> nested_handle.py:12:17
   |
12 |     reveal_type(half)
   |                 ^^^^ `() -> Generator[Ask, Any, None]`

info[revealed-type]: Revealed type
  --> nested_handle.py:13:17
   |
13 |     reveal_type(full)
   |                 ^^^^ `() -> Generator[Never, Any, None]`

half still needs an Ask, and full and nested need nothing, so run() reports a handler left out of either form. ask_tell_stateless.py binds half and full instead of nesting the calls, because a named intermediate is where you read the Ability that remains. That is the information this library exists to give you.

The order of the handlers changes what the checker reports. catch_everything.py supplies first and catches second, and its catch_all(bound) comes back as () -> Generator[Unknown, Any, ...]. With every Ability supplied and every error caught, the channel that should read Never reads Unknown instead. Catching first and supplying second, supply(feed, book)(catch_all(research)), reads Never. That Unknown hides no unsupplied Need. A Need left unsupplied stays named through catch_all(), run() rejects the Effect with an invalid-argument-type, and a run raises a MissingAbilityError naming the same Need. Unknown in a revealed type is still ty not finishing the subtraction, and reveal_type() on the intermediate shows which channel it left unfinished.

The type checker verifies the subtraction, including where you answer part of what an Effect declares. Here supply() answers one Need of two:

# partial_handling.py
from greeter import Console
from stateless import Depend, Need, need, run, supply
from stateless.errors import MissingAbilityError

class Log:
    def write(self, entry: str) -> None:
        print(f"log: {entry}")

def work() -> Depend[Need[Console] | Need[Log], None]:
    console = yield from need(Console)
    log = yield from need(Log)
    console.print("working")
    log.write("worked")

half = supply(Console())(work)
try:
    run(half())  # type: ignore
except MissingAbilityError as e:
    print(e)
#: Need(t=<class '__main__.Log'>)

half is () -> Depend[Need[Log], None], because the handler subtracts the Console and leaves the Log. run() therefore rejects it before the program starts:

error[invalid-argument-type]: Argument to function `run` is incorrect
  --> partial_handling.py:18:9
   |
18 |     run(half())
   |         ^^^^^^ Expected
   |         `Generator[Async | Exception, Any, Unknown]`, found
   |         `Generator[Need[Log], Any, None]`

The # type: ignore silences that diagnostic, since the listing provokes the matching runtime failure on purpose, and the MissingAbilityError it prints names the Log nobody supplied. Nesting the supply() inside the run() call keeps the diagnostic. catch() behaves the same way. If you catch one of two declared errors, the other stays in the error channel.

The library’s types ask the type checker a hard inference question, and the answer changes from one release of the checker to the next. Trust a green check only where you have seen the same construct produce a red one.

3. Handlers cannot capture the continuation

Effect is a monad. success() lifts a value into it, yield from chains two of them together, and the generator body is syntax that hides the chaining. Result in Error Handling has the same two operations, written out by hand. Stateless’s documentation calls the library an algebraic effect system, and both descriptions are right. A monad plus handlers is how you build algebraic effects in a language with no native support for them. The monad is the mechanism that chains the steps, and the handlers are the interface.

What a library cannot copy is the handler’s power. handle() passes a handler the Ability and takes back an answer, and the driver resumes the Effect with that answer, once, always. A native handler instead receives the continuation and chooses what to do with it. Invoking it once gives you what Stateless has. Declining to invoke it makes the handled scope produce the handler’s value instead, which is how an exception behaves. Invoking it repeatedly gives you backtracking and search. Stateless offers only the first of the three, one invocation: a tail-resumptive handler. The limit comes from Python rather than from the library’s design. A Python generator is one-shot, so a handler has nothing to resume twice.

Two pieces of the library show that limit. Effect[A, E, R] carries a separate E that @throws fills and catch() empties. Koka needs no such type parameter, because an exception there is an ordinary Effect whose handler declines to resume. The extra type parameter exists because a Stateless Ability cannot fail, and the ZIO[R, E, A] of Effect Management carries one for the same reason. Async is the other piece. Native systems derive asynchronous execution from Effects. Stateless provides Async as a built-in that run() interprets, because the driver loop can await where a handler cannot.

4. The discipline is all-or-nothing

Every effectful function becomes a generator function, so it cannot also be a plain function. Calling it returns a description that somebody must run. Type errors from a library this generic are long and mention internals. And a third-party function that returns a value or raises an exception needs a @throws wrapper or a need() route before it can participate. An EMS is a decision about a whole codebase, not a utility you import for one module.

5. Much of a mature Effect system is missing

Dependency wiring is the first omission, and bakery.py in Dependencies That Need Dependencies shows its shape. supply() binds finished instances, and handle() takes an ordinary function, so constructing a dependency cannot be an Effect. ZIO’s ZLayer is a constructor that can read configuration, fail, and retry. It resolves a dependency graph at compile time, reporting a cycle or a missing provider by name. Stateless has no equivalent, so you write the wiring at the edge by hand. The type checker verifies that a supply() call is complete but says nothing about how you assembled the graph.

The operator set is small in the same way. The library has retry() and repeat(). Schedule offers a regular interval and a repeat count, with no exponential backoff and no jitter. retry() retries every declared error alike, with no way to name the one worth retrying. Stateless provides no timeout, race, fallback combinator, or finalizer, and the missing race rules out the hedging strategy that races a delayed second request. Concurrency is fork() and wait() with no guarded mutable cell. Forking two Effects that share the Cell of State as an Ability produces a race no type reports, so shared state between forked Effects is your problem and Python’s. Above that sit the resilience patterns a production system eventually needs (rate limiting, bulkheads, and circuit breakers), none of which exist here. The library is a working demonstration of Effect tracking in Python’s type system, and that is different from a platform for building distributed systems.

6. fork() drops the error channel

Running Effects in Parallel names the one restriction the type checker enforces on a forked function: nothing left to supply. The checker enforces nothing about how that function can fail. Two of fork()’s four overloads accept an Effect that still declares an error, and every one of the four returns a Task with no error type on it:

# fork_leak.py
from concurrent.futures import Executor, ThreadPoolExecutor
from exceptions import expect
from stateless import (Async, Depend, Need, Try, as_type,
                       fork, run, supply, throw, wait)

class Boom(Exception):
    pass

@fork
def bad(n: int) -> Try[Boom, int]:
    yield from throw(Boom(n))
    return 0

def go() -> Depend[Need[Executor] | Async, int]:
    task = yield from bad(1)
    value = yield from wait(task)
    return value

with ThreadPoolExecutor(max_workers=1) as pool:
    supplied = supply(as_type(Executor)(pool))(go)
    expect(Boom, run, supplied())
#: [Boom] 1

fork() runs the wrapped Effect with run() inside the worker thread, with no try/except around that call, so a raised failure crosses the thread boundary as an ordinary exception. wait() raises it again when it awaits the Task’s future, past any catch() the caller wraps around the result. The type agrees with the runtime. reveal_type(bad) under ty 0.0.84 reports (n: int) -> Generator[Need[Executor], Any, Task[int]], with Boom nowhere in it.

The fix is the discipline catch() and catch_all() teach: move the failure into the result before you fork. Apply catch_all() before @fork. With bad left undecorated, fork(catch_all(bad)) matches the overload for an Effect with no declared error, and wait() returns Boom | int instead of raising the Boom.

What Survives the Library

Effect Management argues that Effects are the next scaling barrier, and that the tracking will eventually move into the language. Stateless shows what that looks like inside Python today, and that is the value of studying the library, whether or not you use it in production. Effect Tracking looks at the other route Python offers toward algebraic effect tracking: a row written in typing.Annotated, and what a tool would need to check it.

Consider the signatures once more:

• Success[int]: pure.
• Depend[Need[Console], None]: prints, somehow, through something supplied later.
• Effect[Need[Console], KeyError, None]: prints, might not find the name.

Each one describes what a function needs, what it can produce, and how it can fail, before you read a single line of the body. That is the property this book has returned to since Foundations. Purity is valuable because it is verifiable, and verifying it by reading code does not scale.

A second gain shows when you put functions together. Python has a separate mechanism for each concern an Effect type carries. Absence is T | None. Failure is a raised exception, or the Result that Error Handling builds. Asynchrony is async def and await. A resource’s lifetime is a with block. Each is reasonable alone, and none composes with another. Some pairs allow no conversion. An Awaitable cannot become a Result without blocking and giving up the asynchrony. A with block’s guarantee is lexical, so you cannot hand it to a caller as a value the way you can a Result. Consider a function that awaits, might fail, and holds a resource. It uses three of these mechanisms, and its return type mentions one.

Effect[A, E, R] is one type for the dependency, the failure, and the result, and yield from is one operator for joining two Effects. research() joins five steps of two kinds with that one operator, once @throws has brought the ordinary functions in at the boundary. Async is one more Ability in the same channel rather than a second viral annotation. Resource lifetime is the one concern left outside the Effect type. Because Stateless has no scoping mechanism, with blocks stay where they are. ExitStack flattens them, and they remain a separate mechanism from the Effect type.

Both gains demand the generator discipline, the description/execution split, and a wrapper at the boundary for every library that does neither. For most Python code that demand is larger than the benefit. Converting Effectful to Pure gives three techniques: returning a Result, catching the exception you expect, and restricting a type so bad values cannot exist. They make most of a function’s failures visible in its signature, with no generator discipline and no description/execution split. Use Stateless when a system is large enough that hidden Effects have caused a production incident, and when the team will declare every Effect at every boundary. Below that scale, the discipline matters and the machinery is optional.

Whatever you decide about the library, the habit survives it. Name each contact with the outside (the clock, the feed, the pool, the console) and bind it at the edge instead of calling it in the middle. at() and crossing and controller() are all that habit, and none of them needs an Effect type to work.

But watch the direction. Python got one Effect tracked into its type system with async. The languages listed under Native Effect Management track all of them. Stateless demonstrates that Python’s type system is expressive enough to do it, given a library willing to encode everything into return types. What is missing is not the capacity. It is a language that does the encoding for you.

Exercises

This chapter’s solutions give a hint, usually the shape of the code, and a full answer for each exercise.

1. crossing in midnight.py walks a scripted list, so it answers two requests. Write a handler that instead advances a stored moment by one second at each request, and confirm archive() still crosses midnight under it. Then rewrite archive() so the file name and the stamp cannot disagree, and explain why no handler can reproduce the bug afterward.
2. leaky_effect.py type-checks even though its Success[int] claim is false. Describe a review rule or a lint check that catches it, and explain why a type checker cannot. Then demonstrate the error-side twin: write a function that raises a KeyError with no @throws, wrap it in catch(KeyError), and run it on a failing input. Explain what the types claim, what the run does, and which line restores the guarantee.
3. Add a wind turbine to power.py that is available only during a set windy stretch of the evening, put it between solar and the battery in the sun_first order, and confirm run_load() needs no change. Then shorten every source until some hour has no source, run run_load() again, and say where the Blackout propagates to and why catch(Blackout) around run_load() does not intercept it.
4. Write a handler for Outlet that ignores request.hour and hands out a preset sequence of sources, the way scripted hands out a preset sequence of tosses. Use it to test that run_load() re-requests after a failure, without modeling weather, a clock, or a battery. Then say what such a test cannot tell you about controller().
5. Add a fourth failure to research(): a TooLong raised when an article exceeds some length. Follow the type checker’s diagnostics until the program type-checks again, and list every line you edited. Then do the same to research_by_hand.py and say which tool named the lines to change in each case.
6. scenarios.py supplies a DeadWire that fails without printing anything. Write a StaleWire whose latest() prints feed: fetching and then raises Unavailable. Predict the trace, then say why it differs from DeadWire’s even though both fail the same way.
7. Wrap research() in retry() and supply a Time(). Explain what retry() does under the WEATHER scenario and why retrying a NotInteresting failure is the wrong behavior, then say what an Effect system needs for you to retry Unavailable alone.
8. Change parallel.py to use a ProcessPoolExecutor instead of a ThreadPoolExecutor, and confirm squares() stays unchanged. Processes re-import the module, so the driver needs the if __name__ == "__main__": guard Concurrency describes. Without it the pool breaks before any work starts. Then try to fork an Effect that still declares a Need, and record what the type checker says.
9. test_wallet.py runs spree() against a Cell. Script it instead. Write a Get handler that answers from a preset sequence of balances and a Put handler that appends every request to a list, the way scripted feeds Flip. Assert that spree() attempts every price and writes once per purchase. Then say what this test cannot detect that the Cell version can.
10. fetch_nonempty() puts Empty into the channel with throw(). Rewrite it to raise Empty in the body and lift it with @throws(Empty), and confirm the two versions type-check and behave identically. Then make each version fail with an undeclared exception type and compare what the type checker reports for each.
11. Exercise 5 adds a TooLong failure to research(). Repeat it with catch_everything.py in the build. Predict what the type checker reports in outcome(), then confirm. Remove outcome()’s return annotation and rerun ty, and explain what the type checker stopped verifying.
12. Write a Random Ability whose handler returns an int in a range carried on the request, and an accessor roll(low, high) for it. Use it to write a dice game as an Effect, then run the game twice: once with a handler that calls random.randint(), and once with a handler that walks a scripted sequence. Then delete the low: int annotation from the accessor’s parameter and say what changes, and delete the annotation on the handler’s parameter and say what changes.
13. Add a Butter appliance to bakery.py and a buttered() Effect that needs it and calls toast(). Write buttered()’s signature with Need[Butter] alone first, run ty, and read the diagnostic before fixing it. Then remove Toaster(3) from supply() and say which of the two diagnostics tells you about a dependency two levels down.
14. play() in casts.py accepts any three actors, matched or not. kitties_and_puzzles() and warriors_and_weapons() already share a signature. Give that shape a name so a caller can pass either one where a cast belongs, and say what that recovers of the Abstract Factory and what it does not. Then add a fourth actor to encounter() and count the lines you edit in quest.py, casts.py, and two_games.py.