//// A session's run bookkeeping as plain values: what a submission may do,
//// whether a late worker message still belongs to the active run, and which
//// stage a finished run leaves behind. The session actor performs every
//// effect but the cancel latch; these functions only decide, so the rules
//// are testable without a kernel, a provider, or an actor.
import albedo/daemon/conversation
import albedo/daemon/mail
import albedo/openai_api/types
import gleam/dynamic.{type Dynamic}
import gleam/erlang/process
import gleam/list
import gleam/option.{type Option, None, Some}
import gleam/string
/// Who submitted a turn. Chat messages wait in the queue while a run is
/// active; a job wake that finds the session busy is refused so the kernel
/// retries it; a note always waits, and never starts a turn on its own.
pub type Source {
Chat
JobWake
/// A letter from the durable inbox, marked delivered when its input commits.
/// Letters between agents steer a running turn; webhooks wait for idle.
Mail(id: String, kind: mail.Kind)
/// Something an extension tells the agent, such as a user's work ledger
/// change. `origin` labels it in the transcript.
Note(origin: String)
Continue
}
pub type Submission {
Submission(
/// What the transcript view shows; an activation may display less than it sends.
display: String,
/// What the model receives.
text: String,
client_id: String,
source: Source,
image: Option(types.Image),
submission_id: Option(String),
operation_id: Option(String),
)
}
pub type Work {
Compaction
/// An extension's background call: commits nothing and publishes
/// nothing; its outcome goes back to `reply`.
Background(reply: process.Subject(Result(Option(types.Usage), String)))
/// `stage` is None until the turn's first commit.
Turn(stage: Option(conversation.Stage))
}
pub type Run {
Run(
id: String,
pid: process.Pid,
monitor: process.Monitor,
cancelled: Bool,
/// The worker's copy of `cancelled`, raised with it.
stop: Latch,
work: Work,
)
}
/// A flag the session raises once and a worker reads without asking. A call
/// to a stalled session actor gets no answer at all, so a worker that must
/// not start a tool after a cancel cannot learn of it from a reply.
pub type Latch
@external(erlang, "atomics", "new")
fn new_atomics(size: Int, options: List(Nil)) -> Latch
@external(erlang, "atomics", "put")
fn put(latch: Latch, index: Int, value: Int) -> Dynamic
@external(erlang, "atomics", "get")
fn get(latch: Latch, index: Int) -> Int
pub fn latch() -> Latch {
new_atomics(1, [])
}
fn raise(latch: Latch) -> Nil {
let _ = put(latch, 1, 1)
Nil
}
pub fn raised(latch: Latch) -> Bool {
get(latch, 1) == 1
}
pub type Activity {
Resting
/// Idle after a restart that found an unfinished turn it did not resume.
Interrupted
Running(Run)
}
pub type Admission {
Start
Queue
Reject(Rejection)
}
pub type Rejection {
Busy
Oversized
}
/// The most inputs waiting in one session's queue.
const queue_limit = 32
pub fn source_name(source: Source) -> String {
case source {
Chat -> "chat"
JobWake -> "job"
Mail(_, mail.Webhook) -> "webhook"
Mail(..) -> "mail"
Note(origin) -> origin
Continue -> "continue"
}
}
type SourceRule {
StartsTurn
SteersOnly
RefusedWhenBusy
}
fn source_rule(source: Source) -> SourceRule {
case source {
JobWake | Mail(_, mail.Webhook) -> RefusedWhenBusy
Note(_) -> SteersOnly
_ -> StartsTurn
}
}
pub fn admit(
activity: Activity,
submission: Submission,
queued: Int,
) -> Admission {
case
bounded(submission),
queued < queue_limit,
source_rule(submission.source),
activity
{
False, _, _, _ -> Reject(Oversized)
True, False, _, _ -> Reject(Busy)
True, True, SteersOnly, _ -> Queue
True, True, RefusedWhenBusy, Running(_) -> Reject(Busy)
True, True, StartsTurn, Running(_) -> Queue
True, True, _, _ -> Start
}
}
/// A nonempty prompt within its size limits. An activation whose display
/// differs from its text may send a larger body.
fn bounded(submission: Submission) -> Bool {
let maximum = case submission.display == submission.text {
True -> 1_048_576
False -> 2_200_000
}
string.trim(submission.text) != ""
&& string.byte_size(submission.text) <= maximum
&& string.byte_size(submission.display) <= 1_048_576
}
/// Whether a finished run should start another for this queue: notes alone
/// wait for the user's next message.
pub fn starts_turn(queued: List(Submission)) -> Bool {
list.any(queued, fn(submission) {
case submission.source {
Chat | Continue | Mail(..) -> True
JobWake | Note(_) -> False
}
})
}
/// The letters these submissions deliver, for the commit that writes them.
pub fn letters(submissions: List(Submission)) -> List(String) {
list.filter_map(submissions, fn(submission) {
case submission.source {
Mail(id, _) -> Ok(id)
_ -> Error(Nil)
}
})
}
/// Whether letter `id` already waits in this queue.
pub fn holds_letter(queued: List(Submission), id: String) -> Bool {
list.contains(letters(queued), id)
}
pub fn running(activity: Activity) -> Option(Run) {
case activity {
Running(run) -> Some(run)
Resting | Interrupted -> None
}
}
/// The active run, when `id` names it. Durable writes from a cancelled run
/// still land: its completed tool results happened.
pub fn owner(activity: Activity, id: String) -> Option(Run) {
case activity {
Running(run) if run.id == id -> Some(run)
_ -> None
}
}
/// Whether `id` may still publish or take queued input: its run is active and
/// nobody asked it to stop.
pub fn live(activity: Activity, id: String) -> Bool {
case owner(activity, id) {
Some(run) -> !run.cancelled
None -> False
}
}
/// Also raises the worker's latch, so the two copies cannot disagree.
pub fn cancel(activity: Activity) -> Activity {
case running(activity) {
Some(run) -> {
raise(run.stop)
Running(Run(..run, cancelled: True))
}
None -> activity
}
}
/// Record the stage a run's worker committed.
pub fn committed(
activity: Activity,
id: String,
stage: conversation.Stage,
) -> Activity {
case owner(activity, id) {
Some(Run(work: Turn(_), ..) as run) ->
Running(Run(..run, work: Turn(Some(stage))))
_ -> activity
}
}
/// The stage a finished run leaves in the ledger.
pub fn final_stage(run: Run, outcome: Result(a, e)) -> conversation.Stage {
case outcome, run.cancelled {
Ok(_), False -> conversation.Idle
_, _ -> conversation.Interrupted
}
}
/// The phase name clients see in the status route.
pub fn phase(activity: Activity) -> String {
case activity {
Resting -> "resting"
Interrupted -> "interrupted"
Running(Run(work: Compaction, ..)) -> "compacting"
Running(Run(work: Background(_), ..)) -> "background"
Running(Run(work: Turn(None), ..)) -> "preparing"
Running(Run(work: Turn(Some(stage)), ..)) -> conversation.stage_name(stage)
}
}
/// Covered user operations carried by these durable inputs.
pub fn operations(submissions: List(Submission)) -> List(String) {
list.filter_map(submissions, fn(submission) {
option.to_result(submission.operation_id, Nil)
})
}