Actors

An actor owns a state value and changes it one message at a time. Its messages are an indexed variant: each case says whether it is a notification or a request, and a request names the type of its response. One handle function receives every message and returns what happens next as data.

type Counter = {
  count: Int;
}

type CounterMessage('kind) =
  | Increment(Int): CounterMessage(Actor.Notify)
  | Get: CounterMessage(Actor.Request(Int))

The state type implements Actor. Message names the message type, and handle answers each case with a builder: Actor.continue carries on with a new state, and Actor.respond also answers the request. Matching a case refines 'kind, so Get must respond with an Int and Increment cannot respond at all.

module behavior Actor(Counter) {
  type Message('kind) = CounterMessage('kind)

  handlehandle: fn(state: Book.Concurrency.Actors.Counter, message: Actor.Message(Book.Concurrency.Actors.Counter, 'a), from: Actor.Responder('a)) -> Actor.Next('a, Book.Concurrency.Actors.Counter) = (statestate: Book.Concurrency.Actors.Counter, messagemessage: Actor.Message(Book.Concurrency.Actors.Counter, 'a), fromfrom: Actor.Responder('a)) =>
    match messagemessage: Actor.Message(Book.Concurrency.Actors.Counter, 'a) {
      | Increment(byby: Int) -> Actor.continuecontinue: pure fn!('state)(state: Book.Concurrency.Actors.Counter) -> Actor.Next(Actor.Notify, Book.Concurrency.Actors.Counter)({ statestate: Book.Concurrency.Actors.Counter with count = statestate: Book.Concurrency.Actors.Counter.countcount: Int + byby: Int; })
      | Get -> Actor.respondrespond: pure fn!('response, 'state)(value: Int, state: Book.Concurrency.Actors.Counter) -> Actor.Next(Actor.Request(Int), Book.Concurrency.Actors.Counter)(statestate: Book.Concurrency.Actors.Counter.countcount: Int, statestate: Book.Concurrency.Actors.Counter)
    };
}

Notify and request

Actor.start runs the actor on a child fiber of its caller. Actor.notify enqueues a notification and returns at once. Actor.request waits for the typed response, but no longer than within on the provided Clock. The actor handles messages in the order they arrive.

let epochepoch: Instant = must Instant.parse_rfc3339parse_rfc3339: pure fn(string: String) -> Instant, exn [ :parse(String) ]("2026-01-01T00:00:00Z""2026-01-01T00:00:00Z": String);

  provide { Clock = Clock.virtualvirtual: pure fn!(start: Instant) -> Service.Provider(Clock)(epochepoch: Instant); } {
    let countercounter: Actor.Ref(Book.Concurrency.Actors.Counter) = Actor.startstart: pure fn!('state: Actor)(initial: Book.Concurrency.Actors.Counter @ global) -> Actor.Ref(Book.Concurrency.Actors.Counter)(Counter { count = 00: Int; });
    Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Counter) @ global, message: Actor.Message(Book.Concurrency.Actors.Counter, Actor.Notify) @ global) -> Unit(countercounter: Actor.Ref(Book.Concurrency.Actors.Counter), Increment(22: Int));
    Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Counter) @ global, message: Actor.Message(Book.Concurrency.Actors.Counter, Actor.Notify) @ global) -> Unit(countercounter: Actor.Ref(Book.Concurrency.Actors.Counter), Increment(33: Int));
    expect ✓must Actor.requestrequest: fn('state: Actor, 'response)(actor: Actor.Ref(Book.Concurrency.Actors.Counter) @ global, message: Actor.Message(Book.Concurrency.Actors.Counter, Actor.Request(Int)) @ global, ~within: Duration @ global) -> Int, exn [ :stopped | :timeout ], eff Clock(countercounter: Actor.Ref(Book.Concurrency.Actors.Counter), Get, within = Duration.secondsseconds: pure fn!(value: Int) -> Duration(11: Int)) == 55: Int;
  };

Replying later

A handler that cannot answer yet keeps from as an Actor.Promise and returns Actor.promise(state). A later message resolves it by piping Actor.resolve onto its own result. The actor commits each new state before it delivers that message's replies.

A promise is a handle into this runtime with no wire form yet, so a state that stores promises opts out of codecs.

type Semaphore !Encode !Decode = {
  free: Int;
  waiting: Vector(Actor.Promise(Unit));
}

type SemaphoreMessage('kind) =
  | Acquire: SemaphoreMessage(Actor.Request(Unit))
  | Release: SemaphoreMessage(Actor.Notify)
  | Waiters: SemaphoreMessage(Actor.Request(Int))

module behavior Actor(Semaphore) {
  type Message('kind) = SemaphoreMessage('kind)

  handlehandle: fn(state: Book.Concurrency.Actors.Semaphore, message: Actor.Message(Book.Concurrency.Actors.Semaphore, 'a), from: Actor.Responder('a)) -> Actor.Next('a, Book.Concurrency.Actors.Semaphore) = (statestate: Book.Concurrency.Actors.Semaphore, messagemessage: Actor.Message(Book.Concurrency.Actors.Semaphore, 'a), fromfrom: Actor.Responder('a)) =>
    match messagemessage: Actor.Message(Book.Concurrency.Actors.Semaphore, 'a) {
      | Acquire when statestate: Book.Concurrency.Actors.Semaphore.freefree: Int > 00: Int -> Actor.respondrespond: pure fn!('response, 'state)(value: Unit, state: Book.Concurrency.Actors.Semaphore) -> Actor.Next(Actor.Request(Unit), Book.Concurrency.Actors.Semaphore)((), { statestate: Book.Concurrency.Actors.Semaphore with free = statestate: Book.Concurrency.Actors.Semaphore.freefree: Int - 11: Int; })
      | Acquire -> {
        let waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit))) = Vector.pushpush: pure fn!('a)(items: Vector(Actor.Responder(Actor.Request(Unit))), item: Actor.Responder(Actor.Request(Unit))) -> Vector(Actor.Responder(Actor.Request(Unit)))(statestate: Book.Concurrency.Actors.Semaphore.waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit))), fromfrom: Actor.Responder('a));
        Actor.promisepromise: pure fn!('response, 'state)(state: Book.Concurrency.Actors.Semaphore) -> Actor.Next(Actor.Request(Unit), Book.Concurrency.Actors.Semaphore)({ statestate: Book.Concurrency.Actors.Semaphore with waiting = waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit))); })
      }
      | Release when Vector.is_emptyis_empty: pure fn!('a)(items: Vector(Actor.Responder(Actor.Request(Unit)))) -> Bool(statestate: Book.Concurrency.Actors.Semaphore.waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit)))) -> Actor.continuecontinue: pure fn!('state)(state: Book.Concurrency.Actors.Semaphore) -> Actor.Next(Actor.Notify, Book.Concurrency.Actors.Semaphore)(
        { statestate: Book.Concurrency.Actors.Semaphore with free = statestate: Book.Concurrency.Actors.Semaphore.freefree: Int + 11: Int; },
      )
      | Release -> {
        let waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit))) = Vector.dropdrop: pure fn!('a)(items: Vector(Actor.Responder(Actor.Request(Unit))), count: Int) -> Vector(Actor.Responder(Actor.Request(Unit)))(statestate: Book.Concurrency.Actors.Semaphore.waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit))), 11: Int);
        Actor.continuecontinue: pure fn!('state)(state: Book.Concurrency.Actors.Semaphore) -> Actor.Next(Actor.Notify, Book.Concurrency.Actors.Semaphore)({ statestate: Book.Concurrency.Actors.Semaphore with waiting = waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit))); })
        |> Actor.resolveresolve: pure fn!('kind, 'state, 'response)(next: Actor.Next(Actor.Notify, Book.Concurrency.Actors.Semaphore), promise: Actor.Responder(Actor.Request(Unit)), value: Unit) -> Actor.Next(Actor.Notify, Book.Concurrency.Actors.Semaphore)(statestate: Book.Concurrency.Actors.Semaphore.waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit)))[00: Int], ())
      }
      | Waiters -> Actor.respondrespond: pure fn!('response, 'state)(value: Int, state: Book.Concurrency.Actors.Semaphore) -> Actor.Next(Actor.Request(Int), Book.Concurrency.Actors.Semaphore)(Vector.lengthlength: pure fn!('a)(items: Vector(Actor.Responder(Actor.Request(Unit)))) -> Int(statestate: Book.Concurrency.Actors.Semaphore.waitingwaiting: Vector(Actor.Responder(Actor.Request(Unit)))), statestate: Book.Concurrency.Actors.Semaphore)
    };
}

type ActorProbe = {
  mut events: Vector(String);
}

let recordrecord: fn(probe: Book.Concurrency.Actors.ActorProbe @ global, event: String @ global) -> Unit = fn(probeprobe: Book.Concurrency.Actors.ActorProbe, eventevent: String) {
  probe.events := Vector.pushpush: pure fn!('a)(items: Vector(String), item: String) -> Vector(String)(probeprobe: Book.Concurrency.Actors.ActorProbe.eventsevents: Vector(String), eventevent: String);
};

  let probeprobe: Book.Concurrency.Actors.ActorProbe: ActorProbe @ global = ActorProbe { events = []; };
  provide { Clock = Clock.virtualvirtual: pure fn!(start: Instant) -> Service.Provider(Clock)(epochepoch: Instant); } {
    let semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore) = Actor.startstart: pure fn!('state: Actor)(initial: Book.Concurrency.Actors.Semaphore @ global) -> Actor.Ref(Book.Concurrency.Actors.Semaphore)(Semaphore { free = 00: Int; waiting = []; });
    let holderholder: Async.Task(Unit) = spawn {
      must Actor.requestrequest: fn('state: Actor, 'response)(actor: Actor.Ref(Book.Concurrency.Actors.Semaphore) @ global, message: Actor.Message(Book.Concurrency.Actors.Semaphore, Actor.Request(Unit)) @ global, ~within: Duration @ global) -> Unit, exn [ :stopped | :timeout ], eff Clock(
        semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore),
        Acquire,
        within = Duration.secondsseconds: pure fn!(value: Int) -> Duration(55: Int),
      );
      recordrecord: fn(probe: Book.Concurrency.Actors.ActorProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Actors.ActorProbe, "acquired""acquired": String);
    };
    Clock.sleepsleep: fn!(duration: Duration) -> Unit, eff Clock(Duration.zerozero: pure fn!() -> Duration());
    expect ✓must Actor.requestrequest: fn('state: Actor, 'response)(actor: Actor.Ref(Book.Concurrency.Actors.Semaphore) @ global, message: Actor.Message(Book.Concurrency.Actors.Semaphore, Actor.Request(Int)) @ global, ~within: Duration @ global) -> Int, exn [ :stopped | :timeout ], eff Clock(
      semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore),
      Waiters,
      within = Duration.secondsseconds: pure fn!(value: Int) -> Duration(11: Int),
    ) == 11: Int;
    recordrecord: fn(probe: Book.Concurrency.Actors.ActorProbe @ global, event: String @ global) -> Unit(probeprobe: Book.Concurrency.Actors.ActorProbe, "releasing""releasing": String);
    Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Semaphore) @ global, message: Actor.Message(Book.Concurrency.Actors.Semaphore, Actor.Notify) @ global) -> Unit(semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore), Release);
    await holderholder: Async.Task(Unit);
    expect ✓must Actor.requestrequest: fn('state: Actor, 'response)(actor: Actor.Ref(Book.Concurrency.Actors.Semaphore) @ global, message: Actor.Message(Book.Concurrency.Actors.Semaphore, Actor.Request(Int)) @ global, ~within: Duration @ global) -> Int, exn [ :stopped | :timeout ], eff Clock(
      semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore),
      Waiters,
      within = Duration.secondsseconds: pure fn!(value: Int) -> Duration(11: Int),
    ) == 00: Int;
  };
  expect ✓probeprobe: Book.Concurrency.Actors.ActorProbe.eventsevents: Vector(String) == ["releasing""releasing": String, "acquired""acquired": String];

Deadlines

A request that gets no response within its deadline fails with :timeout. The deadline is ordinary Clock time, so a virtual clock reaches it without waiting.

  provide { Clock = Clock.virtualvirtual: pure fn!(start: Instant) -> Service.Provider(Clock)(epochepoch: Instant); } {
    let semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore) = Actor.startstart: pure fn!('state: Actor)(initial: Book.Concurrency.Actors.Semaphore @ global) -> Actor.Ref(Book.Concurrency.Actors.Semaphore)(Semaphore { free = 00: Int; waiting = []; });
    let startstart: Clock.MonotonicInstant = Clock.markmark: fn!() -> Clock.MonotonicInstant();
    let outcomeoutcome: String = try {
      Actor.requestrequest: fn('state: Actor, 'response)(actor: Actor.Ref(Book.Concurrency.Actors.Semaphore) @ global, message: Actor.Message(Book.Concurrency.Actors.Semaphore, Actor.Request(Unit)) @ global, ~within: Duration @ global) -> Unit, exn [ :stopped | :timeout ], eff Clock(
        semaphoresemaphore: Actor.Ref(Book.Concurrency.Actors.Semaphore),
        Acquire,
        within = Duration.secondsseconds: pure fn!(value: Int) -> Duration(22: Int),
      );
      "acquired""acquired": String
    } {
      | :timeout -> "timeout""timeout": String
      | :stopped -> "stopped""stopped": String
    };
    expect ✓outcomeoutcome: String == "timeout""timeout": String;
    expect ✓Clock.elapsed_sinceelapsed_since: fn!(mark: Clock.MonotonicInstant) -> Duration(startstart: Clock.MonotonicInstant) == Duration.secondsseconds: pure fn!(value: Int) -> Duration(22: Int);
  };

Stopping

Actor.stop() ends the actor after the current message. The request that stopped it, every queued request, and every later request fail with :stopped; notifications to a stopped actor are dropped. An actor also stops when its owner ends, like any other child fiber.

type Switch = {
  on: Bool;
}

type SwitchMessage('kind) =
  | Toggle: SwitchMessage(Actor.Notify)
  | Shutdown: SwitchMessage(Actor.Notify)
  | Read: SwitchMessage(Actor.Request(Bool))

module behavior Actor(Switch) {
  type Message('kind) = SwitchMessage('kind)

  handlehandle: fn(state: Book.Concurrency.Actors.Switch, message: Actor.Message(Book.Concurrency.Actors.Switch, 'a), from: Actor.Responder('a)) -> Actor.Next('a, Book.Concurrency.Actors.Switch) = (statestate: Book.Concurrency.Actors.Switch, messagemessage: Actor.Message(Book.Concurrency.Actors.Switch, 'a), fromfrom: Actor.Responder('a)) =>
    match messagemessage: Actor.Message(Book.Concurrency.Actors.Switch, 'a) {
      | Toggle -> Actor.continuecontinue: pure fn!('state)(state: Book.Concurrency.Actors.Switch) -> Actor.Next(Actor.Notify, Book.Concurrency.Actors.Switch)({ statestate: Book.Concurrency.Actors.Switch with on = statestate: Book.Concurrency.Actors.Switch.onon: Bool == FalseFalse: Bool; })
      | Shutdown -> Actor.stopstop: pure fn!('kind, 'state)() -> Actor.Next('kind, Book.Concurrency.Actors.Switch)()
      | Read -> Actor.respondrespond: pure fn!('response, 'state)(value: Bool, state: Book.Concurrency.Actors.Switch) -> Actor.Next(Actor.Request(Bool), Book.Concurrency.Actors.Switch)(statestate: Book.Concurrency.Actors.Switch.onon: Bool, statestate: Book.Concurrency.Actors.Switch)
    };
}

let readread: fn(switch: Actor.Ref(Book.Concurrency.Actors.Switch) @ global) -> String = fn(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch): Actor.Ref(Switch)) {
  try {
    if Actor.requestrequest: fn('state: Actor, 'response)(actor: Actor.Ref(Book.Concurrency.Actors.Switch) @ global, message: Actor.Message(Book.Concurrency.Actors.Switch, Actor.Request(Bool)) @ global, ~within: Duration @ global) -> Bool, exn [ :stopped | :timeout ], eff Clock(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch), Read, within = Duration.secondsseconds: pure fn!(value: Int) -> Duration(11: Int)) {
      | "on""on": String
      | "off""off": String
    }
  } {
    | :stopped -> "stopped""stopped": String
    | :timeout -> "timeout""timeout": String
  }
};

  provide { Clock = Clock.virtualvirtual: pure fn!(start: Instant) -> Service.Provider(Clock)(epochepoch: Instant); } {
    let switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch) = Actor.startstart: pure fn!('state: Actor)(initial: Book.Concurrency.Actors.Switch @ global) -> Actor.Ref(Book.Concurrency.Actors.Switch)(Switch { on = FalseFalse: Bool; });
    Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Switch) @ global, message: Actor.Message(Book.Concurrency.Actors.Switch, Actor.Notify) @ global) -> Unit(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch), Toggle);
    expect ✓readread: fn(switch: Actor.Ref(Book.Concurrency.Actors.Switch) @ global) -> String(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch)) == "on""on": String;
    Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Switch) @ global, message: Actor.Message(Book.Concurrency.Actors.Switch, Actor.Notify) @ global) -> Unit(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch), Shutdown);
    Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Switch) @ global, message: Actor.Message(Book.Concurrency.Actors.Switch, Actor.Notify) @ global) -> Unit(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch), Toggle);
    expect ✓readread: fn(switch: Actor.Ref(Book.Concurrency.Actors.Switch) @ global) -> String(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch)) == "stopped""stopped": String;
  };

  provide { Clock = Clock.virtualvirtual: pure fn!(start: Instant) -> Service.Provider(Clock)(epochepoch: Instant); } {
    let ownerowner: Async.Task(Actor.Ref(Book.Concurrency.Actors.Switch)) = spawn {
      let switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch) = Actor.startstart: pure fn!('state: Actor)(initial: Book.Concurrency.Actors.Switch @ global) -> Actor.Ref(Book.Concurrency.Actors.Switch)(Switch { on = FalseFalse: Bool; });
      Actor.notifynotify: pure fn!('state: Actor)(actor: Actor.Ref(Book.Concurrency.Actors.Switch) @ global, message: Actor.Message(Book.Concurrency.Actors.Switch, Actor.Notify) @ global) -> Unit(switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch), Toggle);
      switchswitch: Actor.Ref(Book.Concurrency.Actors.Switch)
    };
    expect ✓readread: fn(switch: Actor.Ref(Book.Concurrency.Actors.Switch) @ global) -> String(await ownerowner: Async.Task(Actor.Ref(Book.Concurrency.Actors.Switch))) == "stopped""stopped": String;
  };

Actor.notify takes only notification cases and Actor.request only request cases, and a handler arm must answer with the builder its case allows. Rejected programs shows those errors.