Supervision

This is the reason to reach for an actor library instead of a dictionary of asyncio.Queues.

An actor that raises does not take the process down, does not leave a queue with no consumer, and does not silently stop being a thing that runs. It fails, and its parent decides what that means. The decision is written where the child is started, by whoever knew what the child was for.

The four decisions

A strategy is declared by wrapping the behavior: Behaviors.supervise(worker()).on_failure(SupervisorStrategy.restart(), on=IOError).

Decision	What happens	When it fits
resume()	the failure is dropped and the actor carries on with its state	the message was bad, the actor is fine
restart()	the behavior is rebuilt from scratch	the state may be wrong, the actor should start again
stop()	the actor stops, and its watchers hear about it	this actor cannot do its job any more
escalate()	the parent fails too, and its own supervisor decides	this actor's failure means the subtree is broken

on= narrows a layer to a class of failure, and layers are checked from the inside out, so a specific rule can sit inside a general one. A failure nobody wrote a rule for is stopped, not restarted: an actor that failed for a reason nobody anticipated is in a state nobody described, and restarting it in a loop turns one bug into a busy one.

What a restart does, exactly

This is the table worth knowing before choosing restart().

	After a restart
The mailbox	kept. Messages queued behind the failure are still delivered
The message that failed	dropped. It is not retried, because it is what broke the actor
The behavior	re-evaluated from the original one, so setup runs again
State in the closure	gone, which is the point
Children	stopped and respawned by the re-run setup
Timers	cancelled
The stash	cleared
Watchers	told nothing. A restart is not a stop
The ref	unchanged, so everyone holding one keeps working

The last two lines are what makes a restart invisible from outside. Senders do not need to know, and nothing has to be re-resolved.

That also means anything a restart must not forget has to live outside the part that gets re-run. In practice: outside setup, or in another actor.

Backoff

Restarting immediately, forever, against a dependency that is down, is a busy loop with extra steps. SupervisorStrategy.backoff waits, and waits longer each time, with jitter so a fleet of actors that failed together does not retry in lockstep.

"""Restarting an actor whose dependency keeps refusing, without thrashing.

Concepts: `Behaviors.supervise(...).on_failure(...)`, `Restart` with
exponential backoff, the restart window, and what happens to messages sent to
an actor that is between incarnations.

The uploader here stands in for anything that talks to a flaky dependency. Its
first two attempts fail and the third works. Restarting immediately would burn
the whole restart window in a millisecond and stop the actor for a fault that
was about to clear, so the strategy waits, and waits longer each time.

While it waits, the actor is absent, not dead. `tell` stays total, its mailbox
keeps filling, and work sent during the window is handled after the new
incarnation starts rather than dropped. On an unbounded mailbox that costs
memory in proportion to the inbound rate times the window, which is why an
actor that backs off usually wants a bounded mailbox.

The second scenario is the other half of the deal. An actor whose failures
never clear uses up its restart window and is stopped. A supervisor that
restarted forever would turn one bug into a busy one.

What to watch in the output: item 1 fails, and items 2 and 3 were sent while
nobody was there to receive them, yet all three are accounted for. The doomed
actor stops itself after its second failure instead of retrying forever.

Run it with `uv run python -m tapio_examples.supervision_backoff`.
"""

import asyncio
from datetime import timedelta

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import (
    ActorContext,
    Backoff,
    PostStop,
    Signal,
    SupervisorStrategy,
)

__all__ = ["Upload", "main"]

FAILING_ATTEMPTS = 2
"""How many attempts the simulated dependency refuses before it recovers."""

BACKOFF = Backoff(
    min_backoff=timedelta(milliseconds=20),
    max_backoff=timedelta(milliseconds=80),
    # No jitter, so the example is reproducible. Keep the default in
    # production. Without jitter, every actor that saw the same dependency
    # fail retries at the same moment, over and over.
    random_factor=0.0,
)


class Upload(Message):
    """One item to send to the flaky dependency."""

    item: int


def uploader(
    lines: list[str],
    attempts: list[int],
    failed: asyncio.Event,
    recovered: asyncio.Event,
) -> Behavior[Upload]:
    """An uploader whose dependency refuses the first two attempts.

    Args:
        lines: Where to record what happened.
        attempts: Every item attempted, across incarnations. It lives outside
            the behavior on purpose. A restart rebuilds the actor's own state,
            and the dependency it talks to does not reset.
        failed: Set after the first failure, so the example can send into the
            backoff window rather than sleeping and hoping.
        recovered: Set once every item has gone through.

    Returns:
        The supervised behavior.
    """

    def build(ctx: ActorContext[Upload]) -> Behavior[Upload]:
        lines.append(f"uploader: incarnation {len(attempts) + 1} ready")

        async def on_upload(message: Upload) -> Behavior[Upload]:
            attempts.append(message.item)
            if len(attempts) <= FAILING_ATTEMPTS:
                lines.append(f"uploader: item {message.item} failed")
                failed.set()
                msg = "the dependency refused the connection"
                raise ConnectionError(msg)
            lines.append(f"uploader: item {message.item} uploaded")
            recovered.set()
            return Behaviors.same()

        return Behaviors.receive_message(on_upload)

    return Behaviors.supervise(Behaviors.setup(build)).on_failure(
        SupervisorStrategy.restart(
            max_restarts=5, window=timedelta(seconds=1), backoff=BACKOFF
        ),
        # Only the failure this actor knows how to survive. Anything else falls
        # through to stop, which is what an unsupervised actor already does.
        on=ConnectionError,
    )


def doomed(lines: list[str], gave_up: asyncio.Event) -> Behavior[Upload]:
    """An actor whose failure never clears, so its restart window runs out."""

    def build(ctx: ActorContext[Upload]) -> Behavior[Upload]:
        async def on_upload(message: Upload) -> Behavior[Upload]:
            lines.append(f"doomed: item {message.item} failed")
            msg = "this one is never going to work"
            raise ConnectionError(msg)

        async def on_signal(
            ctx: ActorContext[Upload], signal: Signal
        ) -> Behavior[Upload]:
            if isinstance(signal, PostStop):
                lines.append("doomed: restart window exhausted, stopped")
                gave_up.set()
            return Behaviors.same()

        return Behaviors.receive_message(on_upload, on_signal=on_signal)

    return Behaviors.supervise(Behaviors.setup(build)).on_failure(
        # One restart per second. A second failure inside that window says the
        # fault is not transient after all.
        SupervisorStrategy.restart(max_restarts=1, window=timedelta(seconds=1)),
        on=ConnectionError,
    )


async def main() -> list[str]:
    """Run the example.

    Returns:
        One line per thing that happened, in order.
    """
    lines: list[str] = []
    attempts: list[int] = []
    failed, recovered, gave_up = (asyncio.Event(), asyncio.Event(), asyncio.Event())

    async with ActorSystem("supervision") as system:
        flaky = system.spawn(
            uploader(lines, attempts, failed, recovered), name="uploader"
        )

        flaky.tell(Upload(item=1))
        await failed.wait()
        # Sent into the backoff window, at an actor that does not currently
        # exist. Neither send raises, and neither message is lost.
        flaky.tell(Upload(item=2))
        flaky.tell(Upload(item=3))
        await recovered.wait()

        unlucky = system.spawn(doomed(lines, gave_up), name="doomed")
        unlucky.tell(Upload(item=4))
        unlucky.tell(Upload(item=5))
        await gave_up.wait()

    for line in lines:
        print(line)
    return lines


if __name__ == "__main__":
    asyncio.run(main())

Two things in that example are worth noticing. Messages that arrive during the backoff window are buffered rather than dropped, so a recovered actor sees the work that piled up while it was waiting. And a restart window that runs out stops the actor for good, which is how a permanently broken dependency stops being retried forever.

Escalation

Sometimes a child's failure means the parent is broken too. escalate() says so, and the parent fails in turn, which its own supervisor then decides about.

"""A failure the actor that hit it cannot fix, handed to the one that can.

Concepts: `SupervisorStrategy.escalate()`, `ChildFailed`, a whole subtree being
rebuilt by its supervisor, and what happens when an escalation runs out of
supervisors.

A worker that cannot parse its input has no way to repair the pipeline it is
part of. Its parent, which built the pipeline, does. Escalating says that:
stop me, and make this your decision. The parent then takes its own decision,
which here is a restart, so the setup runs again and rebuilds every child
rather than only the one that broke.

Escalation is ordinary message flow, not an exception thrown across a task
boundary. The child stops itself and puts a signal on the parent's system
lane, which is why it can be ordered, observed and tested like anything else.

This example shows two more things. An actor outside the restarted subtree is
untouched, because a child failing must never cancel its siblings. That is why
the runtime uses no task group. And an escalation that reaches the guardian
has run out of actors willing to take responsibility, so the system terminates
and re-raises the cause from `when_terminated`. The service embedding tapio
then decides whether to exit or rebuild.

What to watch in the output: the ticker keeps counting across the restart, and
the second scenario ends with the original error, carrying the path it climbed
through.

Run it with `uv run python -m tapio_examples.escalation`.
"""

import asyncio

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import (
    ActorContext,
    ActorRef,
    PostStop,
    PreRestart,
    Signal,
    SupervisorStrategy,
)

__all__ = ["Parse", "Tick", "main"]


class Parse(Message):
    """A line for the worker to parse. An empty one is unparseable."""

    line: str


class Tick(Message):
    """A nudge for the ticker, which is here to keep working throughout."""


def worker(lines: list[str], parsed: asyncio.Event | None = None) -> Behavior[Parse]:
    """A parser that escalates rather than pretending it can recover."""

    def build(ctx: ActorContext[Parse]) -> Behavior[Parse]:
        lines.append("worker: ready")

        async def on_parse(message: Parse) -> Behavior[Parse]:
            if not message.line:
                lines.append("worker: cannot parse an empty line")
                msg = "empty input"
                raise ValueError(msg)
            lines.append(f"worker: parsed {message.line!r}")
            if parsed is not None:
                parsed.set()
            return Behaviors.same()

        async def on_signal(
            ctx: ActorContext[Parse], signal: Signal
        ) -> Behavior[Parse]:
            if isinstance(signal, PostStop):
                lines.append("worker: stopped")
            return Behaviors.same()

        return Behaviors.receive_message(on_parse, on_signal=on_signal)

    return Behaviors.supervise(Behaviors.setup(build)).on_failure(
        SupervisorStrategy.escalate(), on=ValueError
    )


def pipeline(
    lines: list[str],
    workers: list[ActorRef[Parse]],
    rebuilt: asyncio.Event,
    parsed: asyncio.Event,
) -> Behavior[Parse]:
    """A supervisor that builds its subtree in setup, and so rebuilds it on restart."""

    def build(ctx: ActorContext[Parse]) -> Behavior[Parse]:
        lines.append(f"pipeline: building, incarnation {len(workers) + 1}")
        # Spawned in setup, which is what makes this child come back. A child
        # spawned from a message handler would be gone until that message
        # arrives again.
        workers.append(ctx.spawn(worker(lines, parsed), name="worker"))
        if len(workers) > 1:
            rebuilt.set()

        async def on_parse(message: Parse) -> Behavior[Parse]:
            return Behaviors.same()

        async def on_signal(
            ctx: ActorContext[Parse], signal: Signal
        ) -> Behavior[Parse]:
            if isinstance(signal, PreRestart):
                lines.append("pipeline: restarting after the worker escalated")
            return Behaviors.same()

        return Behaviors.receive_message(on_parse, on_signal=on_signal)

    return Behaviors.supervise(Behaviors.setup(build)).on_failure(
        SupervisorStrategy.restart(max_restarts=3), on=ValueError
    )


def ticker(lines: list[str], ticks: list[int]) -> Behavior[Tick]:
    """An actor with no part in any of this, which is the point of it."""

    async def on_tick(ctx: ActorContext[Tick], message: Tick) -> Behavior[Tick]:
        ticks.append(len(ticks) + 1)
        lines.append(f"ticker: tick {len(ticks)}")
        return Behaviors.same()

    return Behaviors.receive(on_tick)


async def subtree_restarted_by_its_supervisor(lines: list[str]) -> None:
    """Run the first scenario: a worker escalates and its parent rebuilds."""
    workers: list[ActorRef[Parse]] = []
    ticks: list[int] = []
    rebuilt, parsed = asyncio.Event(), asyncio.Event()

    async with ActorSystem("escalation") as system:
        system.spawn(pipeline(lines, workers, rebuilt, parsed), name="pipeline")
        beat = system.spawn(ticker(lines, ticks), name="ticker")

        beat.tell(Tick())
        workers[0].tell(Parse(line=""))
        await rebuilt.wait()

        # The sibling never noticed. A failing actor stops only itself and,
        # through its supervisor's decision, that supervisor's subtree.
        beat.tell(Tick())
        workers[1].tell(Parse(line="ok"))
        await parsed.wait()


async def escalation_that_nobody_catches(lines: list[str]) -> None:
    """Run the second scenario: the escalation reaches the guardian."""
    system = ActorSystem("unsupervised")
    lonely = system.spawn(worker(lines), name="worker")
    lonely.tell(Parse(line=""))

    try:
        await system.when_terminated()
    except ValueError as error:
        lines.append(f"system: terminated by {error}")
        for note in getattr(error, "__notes__", []):
            lines.append(f"system: {note}")


async def main() -> list[str]:
    """Run the example.

    Returns:
        One line per thing that happened, in order.
    """
    lines: list[str] = []
    await subtree_restarted_by_its_supervisor(lines)
    await escalation_that_nobody_catches(lines)

    for line in lines:
        print(line)
    return lines


if __name__ == "__main__":
    asyncio.run(main())

The parent hears about it as a ChildFailed signal, carrying which child and what went wrong. When a subtree restarts, the children are stopped and respawned by the re-run setup, while a sibling in a different subtree is untouched: a failure spreads exactly as far as somebody said it should.

An escalation that reaches a guardian has run out of actors willing to take responsibility, so the system terminates and the cause comes back out of when_terminated(). A crash that nobody handled ends the process rather than leaving it running with a hole in it.

Failure is not the same as refusal

A service that answers "no" has not failed. A payment that is declined, a validation that rejects, a peer that refuses a request: those are answers, and they belong in the reply type where the caller has to deal with them.

"""Three steps that must all happen, and the unwinding when one does not.

Concepts: a saga as an actor, `ask` used as a sequence of steps, and
compensation as ordinary messages.

Payment, then inventory, then shipping. There is no transaction across the
three, because there is no database underneath them: they are separate
services, and once payment has taken the money there is no rolling it back by
not committing. What there is instead is a compensating action for each step,
and an actor whose job is to remember which steps have run.

The saga is a good fit for an actor for one reason. A transaction in flight is
state, and the mailbox means that state is touched by one message at a time.
The steps are awaited in turn, so the actor is parked while a service thinks,
and a second order waits in the queue instead of interleaving with the first.
No lock appears anywhere in here, and there is none to forget to take.

What to watch in the output: the last three lines. Shipping refused, and the
saga did not raise, did not retry, and did not stop. It walked back through
exactly the steps that had succeeded, in reverse, and reported a failed order
rather than a half-finished one. The step that never ran is not compensated,
which is why the list of what to undo is built as the saga goes rather than
written out in advance.

Run it with `uv run python -m tapio_examples.order_saga`.
"""

import asyncio
from collections.abc import Callable
from typing import TypeAlias

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorRef

__all__ = [
    "Charge",
    "Order",
    "Outcome",
    "Refund",
    "Release",
    "Reserve",
    "Ship",
    "Stepped",
    "Unship",
    "main",
    "saga",
    "service",
]


class Stepped(Message):
    """What a service answers: whether it did the thing, and what it says."""

    step: str
    ok: bool
    detail: str


class Charge(Message):
    """Take the money."""

    order: str
    reply_to: ActorRef[Stepped]


class Refund(Message):
    """Give the money back. The compensation for `Charge`."""

    order: str


class Reserve(Message):
    """Hold the stock."""

    order: str
    reply_to: ActorRef[Stepped]


class Release(Message):
    """Put the stock back. The compensation for `Reserve`."""

    order: str


class Ship(Message):
    """Send the parcel."""

    order: str
    reply_to: ActorRef[Stepped]


class Unship(Message):
    """Recall the parcel. The compensation for `Ship`."""

    order: str


class Outcome(Message):
    """What became of an order, and what had to be undone to get there."""

    order: str
    ok: bool
    detail: str
    compensated: list[str]


class Order(Message):
    """Asks for the whole thing to happen, and says where the answer goes."""

    order: str
    reply_to: ActorRef[Outcome]


ServiceMessage: TypeAlias = Charge | Refund | Reserve | Release | Ship | Unship
"""Everything a stand-in service accepts: the three steps and their undos.

One behavior stands in for all three services here, so one ref type describes
all three. A real deployment would have three protocols and three refs, and
the saga below would read the same.
"""


Request: TypeAlias = Callable[[ActorRef[Stepped]], ServiceMessage]
"""Builds one step's request, given where the answer should go."""


def service(
    name: str, lines: list[str], *, refuses: bool = False
) -> Behavior[ServiceMessage]:
    """Build a stand-in for one downstream service.

    Args:
        name: What to call it in the output.
        lines: Where to write what happened.
        refuses: Whether it turns work down. A refusal, not a crash: this
            service is working correctly and the answer is no.

    Returns:
        The behavior to spawn.
    """

    async def on_message(message: ServiceMessage) -> Behavior[ServiceMessage]:
        if isinstance(message, Refund | Release | Unship):
            lines.append(f"saga: {name} undid its part of {message.order}")
            return Behaviors.same()
        if refuses:
            message.reply_to.tell(
                Stepped(step=name, ok=False, detail=f"{name} refused")
            )
            return Behaviors.same()
        lines.append(f"saga: {name} did its part of {message.order}")
        message.reply_to.tell(Stepped(step=name, ok=True, detail=f"{name} agreed"))
        return Behaviors.same()

    return Behaviors.receive_message(on_message, msg_type=ServiceMessage)


def saga(
    payments: ActorRef[ServiceMessage],
    inventory: ActorRef[ServiceMessage],
    shipping: ActorRef[ServiceMessage],
    lines: list[str],
) -> Behavior[Order]:
    """Build the actor that runs one order through the three steps.

    Each step is an `ask`, awaited in turn. That parks the saga while a
    service is thinking, which sounds like a cost and is the point: an order
    part-way through is state, and an actor that is awaiting is not reading
    its mailbox, so a second order waits in the queue rather than interleaving
    with the first. There is no lock in here and there is nothing to forget to
    take.

    Args:
        payments: The service that takes the money.
        inventory: The service that holds the stock.
        shipping: The service that sends the parcel.
        lines: Where to write what happened.

    Returns:
        The behavior to spawn.
    """

    async def on_order(message: Order) -> Behavior[Order]:
        # Built as the saga goes, never written out in advance. A step that
        # never ran has nothing to undo, and a step that refused did nothing
        # to undo either.
        done: list[str] = []
        steps: tuple[tuple[str, ActorRef[ServiceMessage], Request], ...] = (
            ("payments", payments, lambda to: Charge(order=message.order, reply_to=to)),
            (
                "inventory",
                inventory,
                lambda to: Reserve(order=message.order, reply_to=to),
            ),
            ("shipping", shipping, lambda to: Ship(order=message.order, reply_to=to)),
        )
        for name, service_ref, request in steps:
            answer = await service_ref.ask(request, expect=Stepped)
            if not answer.ok:
                # Walk back through what succeeded, newest first.
                for step in reversed(done):
                    _compensate(step, message.order, payments, inventory, shipping)
                message.reply_to.tell(
                    Outcome(
                        order=message.order,
                        ok=False,
                        detail=answer.detail,
                        compensated=list(reversed(done)),
                    )
                )
                return Behaviors.same()
            done.append(name)
        message.reply_to.tell(
            Outcome(
                order=message.order,
                ok=True,
                detail="every step agreed",
                compensated=[],
            )
        )
        return Behaviors.same()

    return Behaviors.receive_message(on_order, msg_type=Order)


def _compensate(
    step: str,
    order: str,
    payments: ActorRef[ServiceMessage],
    inventory: ActorRef[ServiceMessage],
    shipping: ActorRef[ServiceMessage],
) -> None:
    """Send one step's compensating message.

    Args:
        step: Which step to undo.
        order: Which order it was for.
        payments: The service that takes the money.
        inventory: The service that holds the stock.
        shipping: The service that sends the parcel.
    """
    if step == "payments":
        payments.tell(Refund(order=order))
    elif step == "inventory":
        inventory.tell(Release(order=order))
    else:
        shipping.tell(Unship(order=order))


async def main() -> list[str]:
    """Run the example.

    Returns:
        The lines the two orders produced, in order.
    """
    lines: list[str] = []
    async with ActorSystem("orders") as system:
        payments = system.spawn(service("payments", lines), "payments")
        inventory = system.spawn(service("inventory", lines), "inventory")
        # The one that says no. Not a crash: this service is working, and the
        # answer is no. Supervision has nothing to decide about a refusal.
        shipping = system.spawn(service("shipping", lines, refuses=True), "shipping")
        desk = system.spawn(saga(payments, inventory, shipping, lines), "saga")

        outcome = await desk.ask(
            lambda reply_to: Order(order="order-1", reply_to=reply_to), expect=Outcome
        )

        lines.append(f"saga: {outcome.order} failed because {outcome.detail}")
        lines.append(f"saga: undone, newest first: {', '.join(outcome.compensated)}")
        lines.append("saga: nothing was left half done, and nothing raised")

    for line in lines:
        print(line)
    return lines


if __name__ == "__main__":
    asyncio.run(main())

Supervision has nothing to decide about a refusal, and the saga above never raises. It compensates, which is a business decision written in the application, not a lifecycle decision written in a strategy.

Supervision and other nodes

Supervision never crosses a link. A remotely spawned actor is supervised by the node running it, and the requester watches it instead. The reasoning is on the remoting page, and the short version is that every restart decision would otherwise be a frame on a link that can go silent halfway through.

Putting it together

"""One actor per user, over a model that sometimes falls over.

Concepts: state, supervision, ask and watch doing one job together, which is
the shape most real services end up in.

A session is an actor per user. Its conversation is ordinary local state, and
it needs no lock, because one actor handles one message at a time. The model
client is its child, supervised by it, so a model call that blows up is
restarted underneath a session that keeps its history.

The registry above them keeps the map from user to session. It watches every
session it starts, so a session that stops is evicted by the `Terminated` that
follows rather than by whoever remembered to clean up. That is the pattern
worth taking away: watching is how a map of live actors stays true, and it
keeps working when the thing that stopped was not the thing that asked.

What to watch in the output: the third and fourth lines. The model crashed
while it was holding a request, so no answer came back and the ask timed out.
That is not a bug to fix with a bigger timeout: a crash is not a reply, and it
never will be. The session asked again, the restarted model answered, and the
turn count shows the session's own state was never touched.

Run it with `uv run python -m tapio_examples.chat_sessions`.
"""

import asyncio
from datetime import timedelta

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import (
    ActorContext,
    ActorRef,
    Signal,
    SupervisorStrategy,
    Terminated,
)
from tapio.errors import AskTimeoutError

__all__ = [
    "Close",
    "Completion",
    "Prompt",
    "Reply",
    "Say",
    "main",
    "model",
    "registry",
    "session",
]

MODEL_TIMEOUT = timedelta(milliseconds=100)
"""How long a session waits for the model before deciding nothing is coming."""

FAILS_ON_CALL = 2
"""Which call to a model client blows up, so the example is a test as well."""


class Completion(Message):
    """What the model answers."""

    text: str


class Prompt(Message):
    """What the model is asked."""

    text: str
    reply_to: ActorRef[Completion]


class Reply(Message):
    """What the user gets back, and how much of the conversation it has seen."""

    user: str
    text: str
    turns: int


class Say(Message):
    """One thing a user said, and where the answer goes."""

    user: str
    text: str
    reply_to: ActorRef[Reply]


class Close(Message):
    """Ends a user's session."""

    user: str


def model() -> Behavior[Prompt]:
    """Build a stand-in for a model client, scripted to fail on one call.

    The call counter sits here rather than inside `build`, so it survives the
    restart. A restart re-runs `setup`, which is exactly what rebuilds state
    an actor should not keep; anything the restart must not forget has to live
    outside the part that is re-run.

    Returns:
        The behavior to spawn.
    """
    calls = [0]

    def build(ctx: ActorContext[Prompt]) -> Behavior[Prompt]:
        async def on_prompt(message: Prompt) -> Behavior[Prompt]:
            calls[0] += 1
            if calls[0] == FAILS_ON_CALL:
                msg = "the model client fell over"
                raise RuntimeError(msg)
            message.reply_to.tell(Completion(text=f"about {message.text!r}, then"))
            return Behaviors.same()

        return Behaviors.receive_message(on_prompt, msg_type=Prompt)

    return Behaviors.setup(build)


def session(user: str, lines: list[str]) -> Behavior[Say | Close]:
    """Build one user's session, with its own model client underneath it.

    The client is a child, so this session is its supervisor and decides what
    a failing model call means. A restart happens here, in process, and
    nothing above this actor hears about it.

    Args:
        user: Whose session this is.
        lines: Where to write what happened.

    Returns:
        The behavior to spawn.
    """

    def build(ctx: ActorContext[Say | Close]) -> Behavior[Say | Close]:
        # Ordinary local state. One actor handles one message at a time, so
        # nothing here needs a lock and nothing can interleave with it.
        history: list[str] = []
        client = ctx.spawn(
            Behaviors.supervise(model()).on_failure(
                SupervisorStrategy.restart(), on=RuntimeError
            ),
            "model",
        )

        async def on_message(message: Say | Close) -> Behavior[Say | Close]:
            if isinstance(message, Close):
                return Behaviors.stopped()
            history.append(message.text)
            try:
                completion = await client.ask(
                    lambda reply_to: Prompt(text=message.text, reply_to=reply_to),
                    expect=Completion,
                    timeout=MODEL_TIMEOUT,
                )
            except AskTimeoutError:
                # The model client crashed while it was holding the request,
                # so nobody ever answered. A crash is not a reply. Its
                # supervisor has restarted it by now, and the ref is the same
                # one, so asking again is the whole of the recovery.
                lines.append(f"chat: no answer for {user}, so ask the new client")
                completion = await client.ask(
                    lambda reply_to: Prompt(text=message.text, reply_to=reply_to),
                    expect=Completion,
                    timeout=MODEL_TIMEOUT,
                )
            message.reply_to.tell(
                Reply(user=user, text=completion.text, turns=len(history))
            )
            return Behaviors.same()

        return Behaviors.receive_message(on_message, msg_type=Say | Close)

    return Behaviors.setup(build)


def registry(lines: list[str]) -> Behavior[Say | Close]:
    """Build the actor that owns the map from user to session.

    Args:
        lines: Where to write what happened.

    Returns:
        The behavior to spawn.
    """

    def build(ctx: ActorContext[Say | Close]) -> Behavior[Say | Close]:
        sessions: dict[str, ActorRef[Say | Close]] = {}
        users: dict[str, str] = {}

        def session_for(user: str) -> ActorRef[Say | Close]:
            existing = sessions.get(user)
            if existing is not None:
                return existing
            started = ctx.spawn(session(user, lines), f"session-{user}")
            # Watched, not just remembered. The eviction below then happens
            # because the session stopped, whatever stopped it, rather than
            # because somebody remembered to tidy up after one particular way
            # of stopping it.
            ctx.watch(started)
            sessions[user] = started
            users[str(started.path)] = user
            lines.append(f"chat: {user} has a session at {started.path.name}")
            return started

        async def on_message(message: Say | Close) -> Behavior[Say | Close]:
            session_for(message.user).tell(message)
            return Behaviors.same()

        async def on_signal(
            ctx: ActorContext[Say | Close], signal: Signal
        ) -> Behavior[Say | Close]:
            if isinstance(signal, Terminated):
                user = users.pop(str(signal.ref.path), None)
                if user is not None:
                    del sessions[user]
                    lines.append(f"chat: {user}'s session stopped, so it is evicted")
            return Behaviors.same()

        return Behaviors.receive_message(
            on_message, msg_type=Say | Close, on_signal=on_signal
        )

    return Behaviors.setup(build)


async def main() -> list[str]:
    """Run the example.

    Returns:
        The lines the chat produced, in the order it produced them.
    """
    lines: list[str] = []
    async with ActorSystem("chat") as system:
        desk = system.spawn(registry(lines), "sessions")

        for user, text in (("alice", "hello"), ("bob", "hi"), ("alice", "again")):
            answer = await desk.ask(
                lambda reply_to, user=user, text=text: Say(  # type: ignore[misc]
                    user=user, text=text, reply_to=reply_to
                ),
                expect=Reply,
            )
            lines.append(
                f"chat: {answer.user} heard {answer.text!r} on turn {answer.turns}"
            )

        desk.tell(Close(user="alice"))
        # The eviction is a signal arriving at the registry, so give it a turn
        # of the loop to be delivered.
        await asyncio.sleep(0.01)
        # Snapshotted here rather than after the block, because shutdown stops
        # bob's session too and the registry evicts that one as well. That is
        # correct and it is not what this example is about.
        lines = list(lines)

    for line in lines:
        print(line)
    return lines


if __name__ == "__main__":
    asyncio.run(main())

A session per user, a model client per session, supervision at the level that knows what the failure means, and a registry that watches. The model crashes while it is holding a request, and the request is simply lost: a crash is not a reply, and no timeout makes it one. The session asks again, the restarted client answers, and the session's own state was never involved.