Keep an actor safe

A @public handler is open to anyone, and the actor pays for the work it starts, so an open actor needs some protection. This page builds a help desk that has it: it decides who may call each handler, limits what one caller can start, can be paused, and keeps its work small. First, install the CLI.

1. Create the project

cowboy init desk --dir desk
cd desk

Output

✓ Project initialized.
→ Preparing Python environment
✓ Python environment ready · .venv (0.1s)

Yeehaw! Project ready: /path/to/desk

2. Write the actor

Replace actors/desk/main.py with:

from cowboy_sdk import (
    HOUR,
    OWNER,
    BoundedRange,
    Pausable,
    bounded_loop,
    callable_by,
    exit,
    limit,
    public,
    runner,
    single_flight,
)

MAX_THREADS = 50


class DeskActor(Pausable):
    @public
    def init(self, payload: bytes | None = None) -> None:
        self.storage["threads"] = {}

    @public
    @limit(per_caller=20, every=HOUR, pending=2)
    @single_flight(key="thread")
    async def ask(self, request: object) -> str:
        if not isinstance(request, dict) or not isinstance(request.get("text"), str):
            raise ValueError('send {"thread": id, "text": question}')
        thread = str(request["thread"])
        who = self.sender.to_hex()  # read the caller before the await
        threads = self.storage["threads"]
        if thread not in threads and len(threads) >= MAX_THREADS:
            raise ValueError("too many threads")
        answer = await runner.llm(request["text"], max_tokens=256)
        threads = self.storage["threads"]
        threads[thread] = {"by": who, "answer": answer.text}
        self.storage["threads"] = threads
        return answer.text

    @exit
    @public
    def forget_mine(self, _payload: object) -> int:
        return self._forget(self.sender.to_hex())

    @callable_by(OWNER)
    def cancel_waits(self, _payload: object) -> int:
        cancelled = 0
        for wait in runner.waits():
            cancelled += runner.cancel(wait["cid"])
        return cancelled

    @bounded_loop(max_iterations=MAX_THREADS)
    def _forget(self, who: str) -> int:
        threads = self.storage["threads"]
        names = list(threads)
        kept = {}
        for i in BoundedRange(len(names)):
            if threads[names[i]]["by"] != who:
                kept[names[i]] = threads[names[i]]
        self.storage["threads"] = kept
        return len(threads) - len(kept)

Who may call each handler

Decorator	Who may call
`@public`	Anyone
`@callable_by(OWNER)`	The actor’s owner: the account that deployed it
`@callable_by(address, ...)`	The addresses you list
`@callable_by(SELF)`	The actor itself: its timers, and its own messages
None	Nobody from outside

A method with no decorator is for internal use, so name it with a leading underscore, such as _forget. A call that isn’t allowed raises PermissionDeniedError and changes nothing. One thing changes after an await. The code there runs when the answer arrives, and self.sender is then the actor itself, so read the caller before the first await, as ask does with who.

Limit what one caller can start

Each job a handler starts is paid from the actor’s balance, so one caller could spend it all. Two decorators cap that. @limit(per_caller=20, every=HOUR, pending=2) lets each caller call ask at most 20 times an hour, with at most 2 questions waiting for an answer at once, and a call past either limit raises LimitReached. @single_flight(key="thread") allows one open question per thread, and a second one raises Busy until the first finishes. The SDK also caps how many questions can wait at once and always leaves room for the owner and callers you name, so a flood can’t shut you out (limits). runner.cancel(cid) ends a wait.

Pause it

Code can’t change after deploy and no CLI command stops an actor, so an actor that may need stopping has to be built for it from the start. It inherits Pausable, and gives people an @exit handler to take out what is theirs. Pausable adds two handlers only its owner can call: pause and unpause. While the actor is paused, only the owner’s calls and handlers marked @exit can change anything, and scheduled runs are skipped. To end a schedule altogether, use every.stop. @exit grants no permission itself, so pair it with @public or @callable_by.

Bound the work

A call that does too much work fails, so keep each one small. Start by capping what you store: ask refuses a new thread past MAX_THREADS, so the dict, and every loop over it, stays small. Loops themselves can be bounded too. @bounded_loop(max_iterations=N) with BoundedRange raises LoopBoundExceeded past N turns. A handler can’t await in a loop, and it has at most 8 awaits (Call models, the web and tools).

3. Test it

Replace tests/test_desk.py with:

import pytest
from cowboy_sdk import Busy, LimitReached, PermissionDeniedError, codec, mock_host
from cowboy_sdk.pausable import Paused
from cowboy_sdk.testing import SimulatedChain

from actors.desk import main

ALICE = b"\x33" * 20
BOB = b"\x44" * 20


def deploy() -> tuple[main.DeskActor, SimulatedChain]:
    actor = main.DeskActor()
    chain = SimulatedChain(actor)
    mock_host.seed_actor_owner(mock_host.context()["sender"])
    chain.call(actor.init)
    return actor, chain


def ask(thread: str, text: str = "Hi?") -> bytes:
    return codec.encode({"thread": thread, "text": text})


def test_one_open_question_per_thread() -> None:
    actor, chain = deploy()
    chain.set_sender(ALICE)
    chain.call(actor.ask, ask("t1"))
    with pytest.raises(Busy):
        chain.call(actor.ask, ask("t1"))
    chain.reply({"text": "Hello"})
    chain.call(actor.ask, ask("t1"))


def test_two_open_questions_per_caller() -> None:
    actor, chain = deploy()
    chain.set_sender(ALICE)
    chain.call(actor.ask, ask("t1"))
    chain.call(actor.ask, ask("t2"))
    with pytest.raises(LimitReached):
        chain.call(actor.ask, ask("t3"))
    chain.set_sender(BOB)
    chain.call(actor.ask, ask("t3"))


def test_twenty_questions_an_hour() -> None:
    actor, chain = deploy()
    chain.set_sender(ALICE)
    for i in range(20):
        chain.call(actor.ask, ask(f"t{i}"))
        chain.reply({"text": "ok"})
    with pytest.raises(LimitReached):
        chain.call(actor.ask, ask("t20"))
    mock_host.set_context(timestamp_ms=mock_host.context()["timestamp_ms"] + 3_600_000)
    chain.call(actor.ask, ask("t20"))


def test_paused_actor_refuses_questions_but_lets_users_leave() -> None:
    actor, chain = deploy()
    chain.set_sender(ALICE)
    chain.call(actor.ask, ask("t1"))
    chain.reply({"text": "ok"})
    chain.set_sender(mock_host.actor_owner())
    chain.call(actor.pause)
    chain.set_sender(ALICE)
    with pytest.raises(Paused):
        chain.call(actor.ask, ask("t2"))
    assert codec.decode(chain.call(actor.forget_mine)) == 1
    chain.set_sender(BOB)
    with pytest.raises(PermissionDeniedError):
        chain.call(actor.unpause)


def test_the_owner_cancels_open_waits() -> None:
    actor, chain = deploy()
    chain.set_sender(ALICE)
    chain.call(actor.ask, ask("t1"))
    chain.set_sender(mock_host.actor_owner())
    assert codec.decode(chain.call(actor.cancel_waits)) == 1
    name, _ = chain.events()[-1]
    assert name == "cowboy.wait.failed"

cowboy check
cowboy test

Output

✓ Checked desk
  …
tests/test_desk.py .....                                                 [100%]
============================== 5 passed in 0.08s ===============================

mock_host.set_context(timestamp_ms=...) moves the clock, here one hour on, so the hourly limit starts over.

Ownership and code changes

The owner is the account that deployed the actor, and cowboy actor transfer-owner hands it to another account. The code can’t change after deploy, so to ship a change, deploy a new actor (Deploy and operate).

Welcome

Hire Actors

Build Actors

Sell in the Store

Protocol and architecture

Reference

Keep an actor safe

1. Create the project

2. Write the actor

Who may call each handler

Limit what one caller can start

Pause it

Bound the work

3. Test it

Ownership and code changes

Next

​1. Create the project

​2. Write the actor

​Who may call each handler

​Limit what one caller can start

​Pause it

​Bound the work

​3. Test it

​Ownership and code changes

​Next

1. Create the project

2. Write the actor

Who may call each handler

Limit what one caller can start

Pause it

Bound the work

3. Test it

Ownership and code changes

Next