The life of an actor

An actor is three things that always travel together: a path, a mailbox, and a behavior. The path is where it sits in the tree, the mailbox is what it has been sent, and the behavior is what it does with the next message. Everything on this page follows from those three.

Starting

system.spawn(behavior, name) starts a top-level actor, and ctx.spawn from inside a behavior starts a child. Both return a ref, immediately: the actor starts on its own, nobody waits for it, and a message sent to it in the next line arrives after it has started.

"""The smallest complete actor program: spawn, tell, reply, shut down.

Concepts: starting an `ActorSystem`, spawning a top-level actor, sending it a
message with `tell`, and carrying a return address in the message as an
`ActorRef` field.

There is no `ask` here. A reply is just another message, sent to a ref the
sender put in the request. Seeing that once makes `ask` read as the sugar it
is.

What to watch in the output: the greeter's line comes first, then the
listener's, because the reply is a second message and cannot overtake the
handler that sends it. Both actors stop when the system terminates.

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

import asyncio
from collections.abc import Callable

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

__all__ = ["Greet", "Greeted", "main"]


class Greeted(Message):
    """Sent back to whoever asked for the greeting."""

    whom: str


class Greet(Message):
    """A request to greet someone, carrying the address for the reply."""

    whom: str
    reply_to: ActorRef[Greeted]


def greeter(record: Callable[[str], None]) -> Behavior[Greet]:
    """Build the greeter: it greets, replies, and stays as it is.

    Args:
        record: Where to write the greeting, so the example can be asserted.

    Returns:
        The behavior to spawn.
    """

    async def on_greet(ctx: ActorContext[Greet], message: Greet) -> Behavior[Greet]:
        ctx.log.info("hello, %s!", message.whom)
        record(f"greeter: hello, {message.whom}!")
        message.reply_to.tell(Greeted(whom=message.whom))
        return Behaviors.same()

    return Behaviors.receive(on_greet)


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

    Returns:
        The lines the actors produced, in the order they produced them.
    """
    lines: list[str] = []
    done = asyncio.get_running_loop().create_future()

    async def on_greeted(message: Greeted) -> Behavior[Greeted]:
        lines.append(f"listener: {message.whom} has been greeted")
        # Handing a result back out to non-actor code. A future is the way to
        # do that from inside a handler, and `ask` packages this pattern once
        # the request/response shape is familiar.
        done.set_result(None)
        return Behaviors.same()

    async with ActorSystem("hello") as system:
        listener = system.spawn(Behaviors.receive_message(on_greeted), name="listener")
        hello = system.spawn(greeter(lines.append), name="greeter")

        hello.tell(Greet(whom="world", reply_to=listener))
        await done

    for line in lines:
        print(line)
    return lines


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

Names are unique among siblings, and reusing one raises ActorNameError rather than silently addressing the wrong actor. Where a name does not matter, spawn_anonymous generates one beginning with $.

Every ref carries an incarnation uid as well as a path. A ref to an actor that has stopped does not become a ref to the next actor at that path, which is what stops a stale reference from quietly addressing a stranger.

Holding state

State lives in a closure over Behaviors.setup, or in the fields of an AbstractBehavior. Both are ordinary Python variables, and neither needs a lock, because one actor handles one message at a time and nothing else in the process can be inside its handler.

"""An actor that holds mutable state, written in the class-based style.

Concepts: `AbstractBehavior` for an actor with fields, a union message type,
and answering a query by sending to the `reply_to` address it carried.

The count is an ordinary attribute, changed in place with no lock anywhere.
One actor handles one message at a time, so the mailbox already gives the
mutual exclusion a lock would.

What to watch in the output: the reply reports 3, not 1. Messages sent to one
actor from one place arrive in order, so all three increments are applied
before the query behind them.

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

import asyncio

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

__all__ = ["Count", "Counter", "GetCount", "Increment", "main"]


class Count(Message):
    """The answer to a `GetCount`."""

    value: int


class Increment(Message):
    """Add to the count."""

    by: int = 1


class GetCount(Message):
    """Ask for the count, and say where to send it."""

    reply_to: ActorRef[Count]


class Counter(AbstractBehavior[Increment | GetCount]):
    """Counts, and reports the count when asked.

    The message type is read from the type parameter, so nothing has to repeat
    it. A message of any other type is refused at the sender rather than
    landing in this mailbox.
    """

    def __init__(self, ctx: ActorContext[Increment | GetCount]) -> None:
        """Start at zero."""
        super().__init__(ctx)
        self._count = 0

    async def on_message(
        self, message: Increment | GetCount
    ) -> Behavior[Increment | GetCount]:
        """Apply an increment, or answer a query."""
        match message:
            case Increment(by=by):
                self._count += by
                self.ctx.log.debug("count is now %d", self._count)
            case GetCount(reply_to=reply_to):
                reply_to.tell(Count(value=self._count))
        return Behaviors.same()


async def main() -> int:
    """Run the example.

    Returns:
        The count the actor reported.
    """
    answer: asyncio.Future[int] = asyncio.get_running_loop().create_future()

    async def on_count(message: Count) -> Behavior[Count]:
        answer.set_result(message.value)
        return Behaviors.same()

    async with ActorSystem("counter") as system:
        readout = system.spawn(Behaviors.receive_message(on_count), name="readout")
        # Deferred construction. The factory runs when the actor starts, which
        # is what gives the behavior its context, and a restart runs it again.
        counter = system.spawn(Behaviors.setup(Counter), name="counter")

        counter.tell(Increment())
        counter.tell(Increment(by=2))
        counter.tell(GetCount(reply_to=readout))
        value = await answer

    print(f"counter: {value}")
    return value


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

That is the property to keep in mind when deciding what should be an actor: anything that would otherwise need a mutex is a candidate.

Changing what happens next

A handler returns the behavior for the next message. Behaviors.same() keeps the current one, and returning a different behavior switches to it, which is how a protocol with states is written without a state variable and a chain of ifs:

"""A protocol as a state machine, with one behavior per state.

Concepts: behavior-switching as the states of a protocol, and
`ctx.message_adapter` for talking to a service whose replies are not in your
protocol.

The usual way to write a connection is a field called `state`, an enum, and a
`match` at the top of every handler deciding whether this message is legal
right now. An actor needs none of that. The state is the behavior. A
connection that has not authenticated is a different function from one that
has, so a `Send` arriving too early is not an illegal combination to check
for. It is simply a message that state does not handle.

The gain is that the illegal transitions cannot be written. No branch in
`ready` can accidentally accept a second `Authenticate`, because `ready` never
mentions it. And there is no state variable to leave inconsistent, because
switching state means returning a different behavior and nothing else.

The token service is the second half. It is somebody else's actor and it
answers in its own vocabulary, `Token`, which does not belong in a
connection's protocol. Widening the connection to accept a `Token` would let
anyone send it one. Instead the connection hands out an adapter, which accepts
a `Token` and turns it into the `Authenticated` the connection understands.
The translation runs inside the connection, so a mistake in it is the
connection's failure and not the token service's.

What to watch in the output: the `Send` that arrives while connecting is
refused rather than queued, and the identical `Send` after authentication goes
through. Same message, same actor, different state.

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

import asyncio

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

__all__ = ["Close", "Connect", "Send", "main"]


class Token(Message):
    """What the token service answers with. Not the connection's protocol."""

    value: str


class Issue(Message):
    """What the token service accepts."""

    reply_to: ActorRef[Token]


class Connect(Message):
    """Open the connection, which starts authentication."""


class Authenticated(Message):
    """A token has arrived, translated into the connection's own protocol."""

    token: str


class Send(Message):
    """Send a payload, which only works once the connection is ready."""

    payload: str


class Close(Message):
    """Close the connection, draining what it is already doing."""


Protocol = Connect | Authenticated | Send | Close


def tokens() -> Behavior[Issue]:
    """A service that mints tokens, in its own vocabulary."""

    async def on_issue(message: Issue) -> Behavior[Issue]:
        message.reply_to.tell(Token(value="t-42"))
        return Behaviors.same()

    return Behaviors.receive_message(on_issue)


def connection(
    lines: list[str], marks: dict[int, asyncio.Event], service: ActorRef[Issue]
) -> Behavior[Protocol]:
    """A connection whose protocol states are its behaviors."""

    def say(line: str) -> None:
        """Write a line down, and signal the points the script waits for."""
        lines.append(line)
        mark = marks.get(len(lines))
        if mark is not None:
            mark.set()

    def build(ctx: ActorContext[Protocol]) -> Behavior[Protocol]:
        # Handed to the token service in place of this actor's own ref, so the
        # service can answer without the connection accepting `Token` into its
        # protocol. The lambda has no annotation, so the type is passed in.
        as_authenticated: ActorRef[Token] = ctx.message_adapter(
            lambda token: Authenticated(token=token.value), Token
        )

        def disconnected() -> Behavior[Protocol]:
            """Nothing is open. The only thing that can happen is `Connect`."""

            async def on_message(message: Protocol) -> Behavior[Protocol]:
                if not isinstance(message, Connect):
                    say(f"conn: refused {type(message).__name__}, not open")
                    return Behaviors.same()
                say("conn: connecting, asking for a token")
                service.tell(Issue(reply_to=as_authenticated))
                return authenticating()

            return Behaviors.receive_message(on_message)

        def authenticating() -> Behavior[Protocol]:
            """Waiting for a token. Still not a state that can send anything."""

            async def on_message(message: Protocol) -> Behavior[Protocol]:
                if not isinstance(message, Authenticated):
                    say(f"conn: refused {type(message).__name__}, still connecting")
                    return Behaviors.same()
                say(f"conn: authenticated with {message.token}")
                return ready(message.token)

            return Behaviors.receive_message(on_message)

        def ready(token: str) -> Behavior[Protocol]:
            """Open. This is the only state that mentions `Send` at all."""

            async def on_message(message: Protocol) -> Behavior[Protocol]:
                match message:
                    case Send(payload=payload):
                        say(f"conn: sent {payload!r} with {token}")
                    case Close():
                        # A marker message. Everything already queued sits
                        # ahead of a message sent now, so when this one comes
                        # back the mailbox is drained and it is safe to stop.
                        say("conn: closing, draining what is queued")
                        ctx.self_ref.tell(Close())
                        return closing()
                    case _:
                        say(f"conn: refused {type(message).__name__}, open")
                return Behaviors.same()

            return Behaviors.receive_message(on_message)

        def closing() -> Behavior[Protocol]:
            """Going away. What was already queued is answered, then it stops."""

            async def on_message(message: Protocol) -> Behavior[Protocol]:
                if isinstance(message, Close):
                    say("conn: closed")
                    return Behaviors.stopped()
                say(f"conn: dropped {type(message).__name__}, closing")
                return Behaviors.same()

            return Behaviors.receive_message(on_message)

        return disconnected()

    return Behaviors.setup(build)


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

    Returns:
        One line per thing the connection did, in order.
    """
    lines: list[str] = []
    marks = {n: asyncio.Event() for n in (3, 4, 8)}

    async with ActorSystem("state-machine") as system:
        service = system.spawn(tokens(), name="tokens")
        conn = system.spawn(connection(lines, marks, service), name="conn")

        # Too early. The connection is not open, and this state does not
        # handle a Send at all.
        conn.tell(Send(payload="hello"))
        conn.tell(Connect())
        # Also too early, and refused by a different state for its own reason.
        conn.tell(Send(payload="hello again"))
        await marks[3].wait()

        # The token arrives through the adapter, and the connection is ready.
        await marks[4].wait()
        conn.tell(Send(payload="hello"))
        conn.tell(Close())
        conn.tell(Send(payload="too late"))
        await marks[8].wait()

    for line in lines:
        print(line)
    return lines


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

Behaviors.stopped() ends the actor. That is the way to stop one: send a message its behavior answers with stopped(), rather than reaching for the runtime.

Signals

Alongside messages, an actor is told about its own lifecycle on the system lane, which is drained ahead of ordinary traffic:

Signal	When
PostStop	after the last message, whatever stopped it
PreRestart	before a restart replaces the behavior
Terminated	an actor this one watched has stopped
ChildFailed	a child failed and the failure escalated to here

PostStop is where a resource an actor opened is closed. It runs for a stop, a restart's teardown and a shutdown alike, so there is one place to write it rather than three.

Watching

Watching is how one actor finds out that another has stopped, without asking and without polling. A liveness check is out of date as soon as you have the answer; Terminated is not, because it is delivered by the thing that stopped.

"""Keeping a registry honest when the things in it can stop on their own.

Concepts: `ctx.watch`, the `Terminated` signal, and evicting an entry without
leaking it.

A registry of live actors is the first thing most people write, and it comes
with the same bug every time: entries for actors that have since stopped. The
obvious fix, asking a ref whether it is still alive, does not exist in tapio
and would not work if it did. The answer would be out of date as soon as the
caller read it, because the actor can stop in between. Watching turns this
around. Instead of asking, you are told, exactly once, on the system lane, and
the entry is removed where that fact arrives.

The sessions here are not the registry's children. Watching is not parenthood.
It is a one-way subscription to "this actor has stopped", which is the most
one actor can know about another it does not supervise.

What to watch in the output: the registry never checks whether a session is
alive. It is told, and the count afterwards shows the entry is gone.

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

import asyncio

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, ActorRef, Signal, Terminated

__all__ = ["Census", "Close", "Register", "main"]


class Close(Message):
    """Ask a session to end itself."""


class Register(Message):
    """Tell the registry about a session it should keep track of."""

    who: str
    session: ActorRef[Close]


class Census(Message):
    """Ask the registry to report who it still holds."""


def session() -> Behavior[Close]:
    """A session that ends when asked, and says nothing on the way out."""

    async def on_close(message: Close) -> Behavior[Close]:
        return Behaviors.stopped()

    return Behaviors.receive_message(on_close)


def registry(
    lines: list[str], evicted: asyncio.Event, counted: asyncio.Event
) -> Behavior[Register | Census]:
    """A registry of live sessions that evicts by being told, not by asking."""

    def build(ctx: ActorContext[Register | Census]) -> Behavior[Register | Census]:
        live: dict[str, ActorRef[Close]] = {}

        async def on_message(
            message: Register | Census,
        ) -> Behavior[Register | Census]:
            match message:
                case Register(who=who, session=ref):
                    live[who] = ref
                    # One call, and from here on this registry cannot hold a
                    # stale entry for that session.
                    ctx.watch(ref)
                    lines.append(f"registry: registered {who}, holding {len(live)}")
                case Census():
                    lines.append(f"registry: holding {sorted(live)}")
                    counted.set()
            return Behaviors.same()

        async def on_signal(
            ctx: ActorContext[Register | Census], signal: Signal
        ) -> Behavior[Register | Census]:
            if isinstance(signal, Terminated):
                # The signal carries the ref, and a ref knows its own path, so
                # the entry is found without a second lookup table.
                who = signal.ref.path.name
                live.pop(who, None)
                lines.append(f"registry: {who} stopped, holding {len(live)}")
                evicted.set()
            return Behaviors.same()

        return Behaviors.receive_message(on_message, on_signal=on_signal)

    return Behaviors.setup(build)


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

    Returns:
        One line per thing the registry did, in order.
    """
    lines: list[str] = []
    evicted, counted = asyncio.Event(), asyncio.Event()

    async with ActorSystem("death-watch") as system:
        desk = system.spawn(registry(lines, evicted, counted), name="registry")
        ada = system.spawn(session(), name="ada")
        grace = system.spawn(session(), name="grace")
        desk.tell(Register(who="ada", session=ada))
        desk.tell(Register(who="grace", session=grace))

        # The session ends for its own reasons. This is the case a liveness
        # check cannot handle: nobody tells the registry, and it finds out
        # anyway.
        ada.tell(Close())
        await evicted.wait()

        desk.tell(Census())
        await counted.wait()

    for line in lines:
        print(line)
    return lines


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

The pattern that page is really about is eviction. A map of live actors stays true because the map's owner watches what it holds, so an entry is removed by the actor stopping rather than by whoever remembered to clean up.

Stopping the tree

await system.terminate() drains the tree from the leaves up: a parent is not stopped before its children, so a child can still send to its parent while it shuts down. async with ActorSystem(...) does the same thing at the end of the block.

"""Shutting a system down on SIGINT, and stopping the tree in the right order.

Concepts: a real signal handler on the event loop, `system.terminate()`, the
bottom-up drain, and `PostStop` as the place a resource is released.

The signal is real. The handler is installed with `add_signal_handler` and the
process sends itself SIGINT, so this exercises the same wiring a deployed
service uses. Nothing is simulated, which matters, because "we handle SIGINT"
is the claim that turns out to be false the first time a container is stopped.

Shutdown is bottom-up, and the whole tree races one deadline rather than one
per actor. Worst-case shutdown therefore follows `shutdown_timeout` instead of
multiplying by the depth of the tree. Each actor sees `PostStop` after its
children have seen theirs, so a connection pool held by a parent outlives the
children still handing work back to it.

`PostStop` is best effort. An actor still stuck in a handler when the deadline
passes is cancelled and may never see it. Release what must be released there,
but do not make correctness depend on it running.

What to watch in the output: the two connections stop before the pool that
owns them, and the pool's own line is last.

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

import asyncio
import os
import signal

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorContext, PostStop, Signal

__all__ = ["Query", "main"]


class Query(Message):
    """A unit of work for a connection to handle."""

    sql: str


def connection(name: str, lines: list[str], served: asyncio.Event) -> Behavior[Query]:
    """One pooled connection, which reports when it is closed."""

    async def on_query(message: Query) -> Behavior[Query]:
        lines.append(f"{name}: ran {message.sql!r}")
        if name == "conn-2":  # the second of the two, so both have answered
            served.set()
        return Behaviors.same()

    async def on_signal(ctx: ActorContext[Query], sig: Signal) -> Behavior[Query]:
        if isinstance(sig, PostStop):
            lines.append(f"{name}: closed")
        return Behaviors.same()

    return Behaviors.receive_message(on_query, on_signal=on_signal)


def pool(lines: list[str], served: asyncio.Event) -> Behavior[Query]:
    """A pool that owns two connections and outlives both of them."""

    def build(ctx: ActorContext[Query]) -> Behavior[Query]:
        for name in ("conn-1", "conn-2"):
            ctx.spawn(connection(name, lines, served), name=name).tell(
                Query(sql="select 1")
            )

        async def on_query(message: Query) -> Behavior[Query]:
            return Behaviors.same()

        async def on_signal(ctx: ActorContext[Query], sig: Signal) -> Behavior[Query]:
            if isinstance(sig, PostStop):
                lines.append("pool: closed, after every connection in it")
            return Behaviors.same()

        return Behaviors.receive_message(on_query, on_signal=on_signal)

    return Behaviors.setup(build)


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

    Returns:
        One line per thing that happened, in order.
    """
    lines: list[str] = []
    served, interrupted = asyncio.Event(), asyncio.Event()
    loop = asyncio.get_running_loop()

    def on_sigint() -> None:
        # It does nothing but wake the shutdown. A signal handler runs outside
        # every actor, so anything it touched directly would be state no
        # mailbox is protecting.
        lines.append("signal: SIGINT, shutting down")
        interrupted.set()

    loop.add_signal_handler(signal.SIGINT, on_sigint)
    try:
        system = ActorSystem("graceful")
        system.spawn(pool(lines, served), name="pool")
        await served.wait()

        os.kill(os.getpid(), signal.SIGINT)
        await interrupted.wait()
        await system.terminate()
    finally:
        # Put the interpreter's own handler back, so a second Ctrl-C after
        # this example is a KeyboardInterrupt again.
        loop.remove_signal_handler(signal.SIGINT)

    for line in lines:
        print(line)
    return lines


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

Shutdown races one deadline for the whole tree, not one per actor, so the worst case tracks shutdown_timeout rather than multiplying by depth. An actor still inside a handler when the deadline passes is cancelled, and the warning names its path so the slow one is identifiable rather than anonymous.

After shutdown starts, spawn raises ActorSystemTerminating and a tell becomes a dead letter. Neither is silent, and neither leaks a task.

Where messages go when nobody takes them

A tell never blocks and never raises about the recipient, so an undelivered message is not an exception. It becomes a dead letter: an event carrying the message, the intended recipient and a reason.

"""Where a message goes when nobody is there to receive it.

Concepts: the dead letter stream, subscribing to it, and the three ways a
message fails to arrive: a stopped actor, a full bounded mailbox, and a system
that has already shut down.

An `ActorRef` stays a valid handle after its actor dies, so `tell` never
raises about the recipient. An "is it alive?" check would be out of date as
soon as you had it. The price is messages with nowhere to go, and the point of
this example is that tapio accounts for every one of them instead of dropping
them quietly.

This example teaches an absence, and subscribing is what makes that possible.
Without the stream, "the message was dropped" and "this example is broken"
would look the same from outside.

What to watch in the output: three dead letters, each naming the message, the
actor it was addressed to, and why it did not arrive. The reasons differ, and
that difference is the useful part.

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

import asyncio

from tapio import ActorSystem, Behavior, Behaviors, Message
from tapio.actor import ActorRef, DeadLetter, MailboxConfig, OverflowStrategy

__all__ = ["Work", "main"]

CAPACITY = 2
"""How many messages the overloaded worker will hold before shedding."""


class Work(Message):
    """A unit of work, numbered so the output can be followed."""

    item: int


def busy(started: asyncio.Event, release: asyncio.Event) -> Behavior[Work]:
    """A worker that takes one item and then stalls, so its mailbox fills."""

    async def on_work(message: Work) -> Behavior[Work]:
        started.set()
        await release.wait()
        return Behaviors.same()

    return Behaviors.receive_message(on_work)


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

    Returns:
        One line per dead letter, in the order the system produced them.
    """
    lines: list[str] = []

    def record(letter: DeadLetter) -> None:
        lines.append(f"{letter.message!r} -> {letter.recipient} ({letter.reason})")

    system = ActorSystem("dead-letters")
    system.dead_letters.subscribe(record)

    # 1. A stopped actor. The ref is still a good handle and the send is still
    #    legal. There is simply nobody there to receive it.
    departed: ActorRef[Work] = system.spawn(Behaviors.stopped(), name="departed")
    departed.tell(Work(item=1))
    await asyncio.sleep(0)

    # 2. A bounded mailbox that overflows. DROP_OLDEST keeps the newest work
    #    and drops the oldest, which is what you want when only the latest
    #    reading matters. Whichever message it drops is accounted for.
    started, release = asyncio.Event(), asyncio.Event()
    worker = system.spawn(
        busy(started, release),
        name="worker",
        mailbox=MailboxConfig(
            capacity=CAPACITY, on_overflow=OverflowStrategy.DROP_OLDEST
        ),
    )
    worker.tell(Work(item=2))
    await started.wait()  # it is now stalled, holding item 2
    for item in (3, 4, 5):  # 3 and 4 fill the mailbox, 5 pushes 3 out
        worker.tell(Work(item=item))

    release.set()
    await system.terminate()

    # 3. A send after the system has gone. Still not an error, and still not
    #    silent. The reason names the system rather than the actor.
    worker.tell(Work(item=6))
    await asyncio.sleep(0)

    for line in lines:
        print(line)
    return lines


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

Subscribing to that stream is what makes an absence observable. Without it, "the message was dropped" and "the code never ran" look identical from outside, which is why this is a stream rather than only a log line.