Stateless: Solutions

If you fix the annotation but keep a caller that runs greet_all() with no environment, as run(greet_all(["Alice", "Bob"])) does, ty reports an invalid-argument-type at that run() call. The run raises a MissingAbilityError before the first greeting prints. The new annotation hands the requirement to every caller, so the solution wraps greet_all in supply(Console()) before calling run().

Removing the # type: ignore from undeclared_need.py produces:

error[invalid-yield]: Yield expression type does not match annotation
 --> undeclared_need.py:7:20
  |
5 | def greet_all(names: list[str]) -> Success[None]:
  |                                    ------------- Function annotated
  |                                    with yield type `Never` here
6 |     for name in names:
7 |         yield from greet(name)
  |                    ^^^^^^^^^^^ expression of type `Need[Console]`,
  |                    expected `Never`

Success[None] is Effect[Never, Never, None]: an Effect that needs nothing, fails with nothing, and returns nothing. Never in the yield channel means no value of any type may travel there, so a single Need[Console] coming up from greet() contradicts it.

The fix is the annotation, and only the annotation:

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

class Console:
    def print(self, message: str) -> None:
        print(message)

def greet(name: str) -> Depend[Need[Console], None]:
    console = yield from need(Console)
    console.print(f"Hello, {name}!")

def greet_all(names: list[str]) -> Depend[
    Need[Console], None
]:
    for name in names:
        yield from greet(name)

run(supply(Console())(greet_all)(["Alice", "Bob"]))
#: Hello, Alice!
#: Hello, Bob!

Fix the signature, not the body. The body stays as it is: greet_all() did the right thing all along. Its signature described a different function.

What greet_all()’s callers must now declare is the point of the exercise. Before, greet_all() claimed to need nothing, so a caller could run it with no environment: run(greet_all(names)). Now every caller has two options, the same two greet()’s callers have. Supply a Console, ending the requirement, or declare Need[Console] in its own return type and pass the requirement further up. No third option exists, which makes the dependency visible. The requirement appears in the signature of every function between the one that uses the Console and the one that supplies it, and the type checker refuses to let any of them stay silent.

# exercise_3.py
from typing import Final, assert_never, reveal_type
from stateless import Success, Try, catch, throws

RAW: Final[dict[str, str]] = {"Alice": "42", "Bob": "seven"}

@throws(KeyError, ValueError)
def read_score(name: str) -> int:
    text = RAW[name]  # KeyError
    return int(text)  # ValueError

both = catch(KeyError, ValueError)(read_score)
one = catch(KeyError)(read_score)

def all_handled(name: str) -> Success[str]:
    value: int | KeyError | ValueError = yield from both(
        name)
    match value:
        case KeyError():
            return f"{name}: unknown"
        case ValueError():
            return f"{name}: unreadable"
        case int():
            return f"{name}: {value}"
        case _:
            assert_never(value)

def one_unhandled(name: str) -> Try[ValueError, str]:
    value: int | KeyError = yield from one(name)
    match value:
        case KeyError():
            return f"{name}: unknown"
        case int():
            return f"{name}: {value}"
        case _:
            assert_never(value)

if __name__ == "__main__":
    reveal_type(catch(ValueError)(one_unhandled))

ty reveals:

info[revealed-type]: Revealed type
`(name: str) -> Generator[Never, Any, str | ValueError]`

all_handled() is Success[str], which expands to Generator[Never, Any, str]. The revealed type and all_handled()’s type both carry Never in the yield channel, so both agree that nothing can fail from here on. They differ in the return channel: str for all_handled(), str | ValueError for the wrapped one_unhandled().

Consume every caught error. The difference is where the error stops travelling. all_handled() catches both errors, then consumes both in its match, turning each into a sentence and returning a str. No error remains, and the assert_never() proves it: the match covers every case inside the function.

Pass the remaining error up. one_unhandled() catches only the KeyError and consumes that one. The ValueError stays declared, as Try[ValueError, str] says, so it is still in the yield channel when catch(ValueError) wraps one_unhandled(). catch() does not delete an error. It moves the error from the yield channel to the return channel, as a value. So the ValueError leaves the failure channel and reappears beside the str, and a caller needing a bare str has one case left to handle.

Both functions have handled every error read_score() declares. Only all_handled() has interpreted what it caught. catch() turns a failure into a value, and a match turns that value into a result. Skipping the match leaves the caught error sitting in the return type.

# exercise_6.py
from stateless import Depend, Need, handle, need, run

class Console:
    def print(self, message: str) -> None:
        print(message)

class Clock:
    def now(self) -> str:
        return "noon"

def greet(name: str) -> Depend[Need[Console], None]:
    console = yield from need(Console)
    console.print(f"Hello, {name}!")

def stamped(
    name: str,
) -> Depend[Need[Console] | Need[Clock], None]:
    clock = yield from need(Clock)
    console = yield from need(Console)
    console.print(f"[{clock.now()}] Hello, {name}!")

def default(ability: Need[Console]) -> Console:
    print(f"handler answered a request "
          f"for {ability.t.__name__}")
    return ability.t()

defaults = handle(default)
run(defaults(greet)("Alice"))
#: handler answered a request for Console
#: Hello, Alice!
run(defaults(stamped)("Bob"))  # type: ignore
#: handler answered a request for Clock
#: handler answered a request for Console
#: [noon] Hello, Bob!

Build the class the request names. default() names no Console in its body. It reads ability.t, the class the request carries, and calls that class, so default() answers a request by constructing the requested class. That is the other kind of default: default_console.py supplies one prepared instance, and default() builds whatever the request names, on demand.

Run both Effects through one handler. At runtime the handler answered three requests across the two calls, two for Console and one for Clock, although default() annotates its parameter Need[Console]. The type checker believes the handler answers Need[Console] and leaves Need[Clock] open, so the second run() needs a # type: ignore. Without that pragma, the type checker reports a leftover Need[Clock].

handle()’s t = get_origin(t) or t is the evidence. handle() reads the annotation, reduces Need[Console] to its origin, Need, and installs the runtime check isinstance(ability, Need). That check ignores the type argument, so every Need[...] request matches. The type checker reads the same annotation without that reduction, subtracts Need[Console] from the requirements, and leaves Need[Clock] in place.

Neither view is wrong about what it describes. isinstance() cannot test a type argument: Need[Clock] and Need[Console] are the same runtime class, so no runtime check tells the two apart. The annotation is the only place the distinction exists, and handle() uses the annotation for matching but cannot enforce the distinction. A handler like default() genuinely handles more than its type says, and one that assumes ability.t is a Console receives a Clock with nothing to stop it.

# exercise_8.py
from collections.abc import Callable
from functools import partial
from typing import Final
from stateless import (Depend, Need, Success, need,
                       run, supply)

class Console:
    def print(self, message: str) -> None:
        print(message)

def greet(name: str) -> Depend[Need[Console], None]:
    console = yield from need(Console)
    console.print(f"Hello, {name}!")

NAMES: Final[list[str]] = ["Alice", "Bob"]

built: dict[str, Success[None]] = {
    name: supply(Console())(greet)(name) for name in NAMES
}
for effect in built.values():
    run(effect)
#: Hello, Alice!
#: Hello, Bob!
# The same objects, a second time
for effect in built.values():
    run(effect)

def make(name: str) -> Success[None]:
    return supply(Console())(greet)(name)

builders: dict[str, Callable[[], Success[None]]] = {
    name: partial(make, name) for name in NAMES
}
for builder in builders.values():
    run(builder())
#: Hello, Alice!
#: Hello, Bob!
# A fresh Effect each time
for builder in builders.values():
    run(builder())
#: Hello, Alice!
#: Hello, Bob!

Run each stored Effect twice. The first pass over built greets both names. The second prints nothing, and run() returns None for each entry without raising an exception. An Effect is a generator, and a generator runs once. Resuming a finished generator raises StopIteration immediately, which run() reads as “already returned, with no value.” So a spent Effect looks the same as one that succeeded and returned None, and nothing reports the difference.

Build a fresh Effect per run. The dictionary of builders behaves as a reader expects. Each pass calls each entry, each call builds a new generator, and each generator runs its body once. The stored value goes from a description run() consumes once to a recipe a caller can follow as often as it likes.

That difference is why retry() takes a schedule and returns a decorator of type Callable[P, Effect[...]] -> Callable[P, Effect[...]] rather than one of type Effect[...] -> Effect[...]. Retrying means running the same work more than once, and an Effect cannot supply the second run: by the time the first attempt fails, that attempt has run the generator to its end, leaving nothing to resume. retry() needs to build a fresh Effect per attempt, and only the function that builds the Effect can do that. So retry() decorates the function, calls it once per attempt, and hands back a function that takes the same arguments. The Effect the returned function builds has a wider type: retry() adds the clock on which it sleeps and replaces the error with a RetryError.

The same reasoning explains repeat() and memoize(). It also explains why storing Effects in a registry, a queue, or a cache is a mistake that looks fine until something runs an entry twice. Store the function, and apply the arguments where you need the Effect.

# exercise_9.py
import asyncio
from exceptions import expect
from stateless import Async, Depend, run, run_async, wait

async def fetch(url: str) -> str:
    await asyncio.sleep(0.01)
    return f"fetched {url}"

def report(url: str) -> Depend[Async, str]:
    body = yield from wait(fetch(url))
    return f"{body = }, {len(body) = }"

def report_all(urls: list[str]) -> Depend[Async, list[str]]:
    reports: list[str] = []
    for url in urls:
        reports.append((yield from report(url)))
    return reports

async def main() -> None:
    expect(RuntimeError, run, report_all(["a"]))
    for line in await run_async(
            report_all(["a", "b", "c"])):
        print(line)

asyncio.run(main())
#: [RuntimeError] asyncio.run() cannot be called from a
#: running event loop
#: body = 'fetched a', len(body) = 9
#: body = 'fetched b', len(body) = 9
#: body = 'fetched c', len(body) = 9

Relay the request, collect the results. The annotation is Depend[Async, list[str]]. report() needs Async, and yield from passes that requirement straight up, so report_all() needs it too. Three delegations to the same Effect type add nothing new to the channel: Need[Console] | Need[Log] grows because the two requirements differ, and here they do not. Only the return type changes, from one str to a list[str], since report_all() collects the results rather than relaying them.

Drive the Effect from a coroutine. run() raises a RuntimeError, and await run_async(...) works. run(effect) is asyncio.run(run_async(effect)). asyncio.run() refuses to start an event loop inside a running one, so calling run() from main() fails. run_async() is the same driver in coroutine form, so the running loop can await it.

The type checker accepts both calls because both are correctly typed. run() takes an Effect and returns its result. run_async() takes an Effect and returns an awaitable of its result. report_all(["a"]) satisfies either signature, and nothing in the type system records that this call site sits inside a coroutine. Whether an event loop is running is a fact about the moment of the call, not about the types involved, so calling run() inside a coroutine is a mistake the type checker cannot report. The rule is positional rather than type-based: run() at the outermost edge of a synchronous program, run_async() anywhere inside an asynchronous one.

# exercise_10.py
from typing import Final
from exceptions import expected
from stateless import (Effect, Need, need, run, supply,
                       throws)

class Console:
    def print(self, message: str) -> None:
        print(message)

SCORES: Final[dict[str, int]] = {
    "Alice": 42, "Bob": 7, "Cyd": -3}

@throws(KeyError)
def score(name: str) -> int:
    return SCORES[name]

@throws(ValueError)
def format_score(name: str, value: int) -> str:
    if value < 0:
        raise ValueError(
            f"negative score for {name}: {value}")
    return f"{name}: {value}"

def announce(
    name: str,
) -> Effect[Need[Console], KeyError | ValueError, None]:
    value: int = yield from score(name)
    line: str = yield from format_score(name, value)
    console = yield from need(Console)
    console.print(line)

bound = supply(Console())(announce)
for who in ("Alice", "Cyd", "Dana"):
    with expected((KeyError, ValueError)):
        run(bound(who))
#: Alice: 42
#: [ValueError] negative score for Cyd: -3
#: [KeyError] 'Dana'

Lift the helper into an Effect. @throws(ValueError) turns format_score() from a function that raises an exception into an Effect that declares one, so its failure travels as a value in the yield channel instead of unwinding the stack.

Declare the second failure. Following the type checker until the program builds means one edit: widening announce()’s error parameter from KeyError to KeyError | ValueError. line: str and value: int carry annotations by choice, not by demand: yield from on a @throws function produces the declared success type, and naming that type keeps the type checker’s inference pinned.

Surface each failure at run(). Each failure surfaces at run(), and nowhere earlier. Cyd has a score, so the lookup succeeds and format_score() fails. Dana has none, so the lookup fails and format_score() does not run. In both cases the error value travels up through the yield from chain untouched, past announce(), past supply(), to the driver. run() raises it as an ordinary exception because nothing along the way catches it. Declaring a failure is not handling it. The declaration says the failure can arrive, and catch() turns it into a value the program handles.

Deleting ValueError from the annotation gives:

error[invalid-yield]: Yield expression type does not match annotation
  --> exercise_10.py:29:28
   |
27 | ) -> Effect[Need[Console], KeyError, None]:
   |      ------------------------------------- Function annotated with
   |      yield type `Need[Console] | KeyError` here
28 |     value: int = yield from score(name)
29 |     line: str = yield from format_score(name, value)
   |                            ^^^^^^^^^^^^^^^^^^^^^^^^^
   |                            expression of type `ValueError`,
   |                            expected `Need[Console] | KeyError`

The error appears on line 29, the yield from that introduces the undeclared failure, not on the signature and not at the call site. That line is the useful place for the diagnostic. The diagnostic names both the failure that escaped and the delegation through which it escaped, so the fix is either to declare the failure or to catch it, right there.

Three implementations have six orderings, and the prediction is short. supply() scans its arguments and takes the first that satisfies the request, so whichever implementation comes first answers it. Two of the six orderings put Terminal first and send the greeting to the screen, and the remaining four split evenly between Capture and the third implementation. No ordering produces an error, and no ordering produces a warning.

The fix is to stop letting one structural check match three objects:

# exercise_11.py
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable
from stateless import (Depend, Need, as_type, need,
                       run, supply)

@runtime_checkable
class Screen(Protocol):
    def print(self, message: str) -> None: ...

@runtime_checkable
class Recorder(Protocol):
    def record(self, message: str) -> None: ...

@dataclass
class Terminal:
    def print(self, message: str) -> None:
        print(message)

@dataclass
class Capture:
    messages: list[str] = field(default_factory=list)
    def record(self, message: str) -> None:
        self.messages.append(message)

def to_screen(name: str) -> Depend[Need[Screen], None]:
    device = yield from need(Screen)
    device.print(f"Hello, {name}!")

def to_log(name: str) -> Depend[Need[Recorder], None]:
    device = yield from need(Recorder)
    device.record(f"Hello, {name}!")

capture = Capture()
run(supply(as_type(Screen)(Terminal()))(to_screen)("Alice"))
#: Hello, Alice!
run(supply(as_type(Recorder)(capture))(to_log)("Bob"))
print(capture.messages)
#: ['Hello, Bob!']

Give each role its own protocol. The fix renames Capture.print() to record(), gives each method its own Protocol, Screen and Recorder, and splits greet() into one Effect per Protocol. The two Protocols no longer overlap, so neither implementation satisfies both, and each Effect names the Protocol it needs.

That change turns the coin flip into a diagnostic. If you add one more line to the end of the listing, handing to_log the object that prints instead of the one that records, ty rejects it before the program runs:

error[invalid-argument-type]: Argument is incorrect
  --> exercise_11.py:40:30
   |
40 | run(supply(as_type(Recorder)(Terminal()))(to_log)("Carol"))
   |                              ^^^^^^^^^^ Expected `Recorder`, found `Terminal`
info: type `Terminal` is not assignable to protocol `Recorder`
info: └── protocol member `record` is not defined on type `Terminal`

In the second info line, the type checker names the missing method rather than the missing type, and that is what structural typing means: Terminal fails not because of what it is but because of what it does not do.

One limit remains. Distinct method names remove the ambiguity between abilities. They do nothing about two implementations of the same ability. If you add a second recorder, an Audit that also defines record(), supply(capture, audit) is ambiguous again by argument order, with no diagnostic. When Two Implementations Match gives its advice in two halves for that reason. No type can enforce the second half, “supply one implementation per Ability.” Stateless resolves a request by scanning its arguments at runtime, so a duplicate is a fact about the call rather than about the types. ZIO’s compile-time rejection of this case is the difference that section names.