diff --git a/src/albedo/daemon/session_run.gleam b/src/albedo/daemon/session_run.gleam index 726eb89..b711b22 100644 --- a/src/albedo/daemon/session_run.gleam +++ b/src/albedo/daemon/session_run.gleam @@ -71,19 +71,22 @@ pub fn try_call( /// The worker's publish, which also gates every tool call. A timeout only /// says the session actor is busy: keep streaming, the durable commit lands -/// once it catches up, and an interrupt that landed during the stall is still -/// backed by the Abort timer and the kernel interrupt. A dead session actor -/// ends the stream, so no tool starts without an owner to record its result. +/// once it catches up. Once the session has cancelled the run, though, a +/// timeout stops it: the kernel interrupt that backs a late-started tool +/// cannot reach a tool outside the kernel, such as an MCP call. A dead session +/// actor ends the stream, so no tool starts without an owner to record its +/// result. pub fn publish_fn( owner: Subject(message), run_id: String, messages: Messages(message), + stop: turn.Latch, waiting timeout: Int, ) -> fn(String) -> Bool { fn(event) { case try_call(owner, timeout, messages.publish(run_id, event, _)) { Ok(keep_going) -> keep_going - Error(TimedOut) -> True + Error(TimedOut) -> !turn.raised(stop) Error(CalleeDown) -> False } } @@ -162,6 +165,7 @@ pub fn start( ) -> session_state.State(message) { let run_id = new_id() let owner = state.self + let stop = turn.latch() // The worker's closures must capture these fields, never `state`: a spawn // copies everything its closure references, and the session state carries // the loaded transcript. @@ -175,7 +179,7 @@ pub fn start( kernel, state.pin, client, - publish_fn(owner, run_id, messages, 30_000), + publish_fn(owner, run_id, messages, stop, 30_000), commit_fn(owner, run_id, messages, 10_000), fn(request, observation) { let snapshot = @@ -211,7 +215,7 @@ pub fn start( }), ) }) - let run = turn.Run(run_id, pid, process.monitor(pid), False, work) + let run = turn.Run(run_id, pid, process.monitor(pid), False, stop, work) // The worker holds the projected history it runs on; the session's copy is // released until the next load, since every commit is durable first. process.send(owner, messages.collect) diff --git a/src/albedo/daemon/turn.gleam b/src/albedo/daemon/turn.gleam index f4cfb5e..6b2fd87 100644 --- a/src/albedo/daemon/turn.gleam +++ b/src/albedo/daemon/turn.gleam @@ -1,12 +1,13 @@ //// 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; these functions only decide, so the rules are testable without a -//// kernel, a provider, or an actor. +//// 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} @@ -51,10 +52,39 @@ pub type Run { 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, []) +} + +pub 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. @@ -173,9 +203,13 @@ pub fn live(activity: Activity, id: String) -> Bool { } } +/// Also raises the worker's latch, so the two copies cannot disagree. pub fn cancel(activity: Activity) -> Activity { case activity { - Running(run) -> Running(Run(..run, cancelled: True)) + Running(run) -> { + raise(run.stop) + Running(Run(..run, cancelled: True)) + } other -> other } } diff --git a/test/daemon/session_run_test.gleam b/test/daemon/session_run_test.gleam index 1fdb4be..dbef79a 100644 --- a/test/daemon/session_run_test.gleam +++ b/test/daemon/session_run_test.gleam @@ -3,6 +3,7 @@ import albedo/daemon/conversation import albedo/daemon/session_run +import albedo/daemon/turn import albedo/daemon/usage import gleam/erlang/process.{type Subject} import gleam/option.{None} @@ -98,7 +99,11 @@ fn messages() -> session_run.Messages(Owner) { } fn publish(owner: Subject(Owner)) -> Bool { - session_run.publish_fn(owner, "run", messages(), waiting: 20)("event") + publish_after(owner, turn.latch()) +} + +fn publish_after(owner: Subject(Owner), stop: turn.Latch) -> Bool { + session_run.publish_fn(owner, "run", messages(), stop, waiting: 20)("event") } pub fn publish_passes_the_owner_answer_through_test() { @@ -121,6 +126,25 @@ pub fn publish_keeps_streaming_through_a_stalled_owner_test() { |> should.be_true } +/// A cancelled session actor that stalls must not let a tool gate open: the +/// kernel interrupt cannot stop a tool outside the kernel. +pub fn publish_fails_closed_on_a_stall_once_cancelled_test() { + let pid = process.spawn(fn() { Nil }) + let assert turn.Running(run) = + turn.Running(turn.Run( + "run", + pid, + process.monitor(pid), + False, + turn.latch(), + turn.Turn(None), + )) + |> turn.cancel + server(fn(_) { actor.continue(Nil) }) + |> publish_after(run.stop) + |> should.be_false +} + pub fn publish_stops_the_stream_when_the_owner_dies_test() { server(fn(_) { actor.stop() }) |> publish diff --git a/test/daemon/turn_test.gleam b/test/daemon/turn_test.gleam index 2170513..6638c3c 100644 --- a/test/daemon/turn_test.gleam +++ b/test/daemon/turn_test.gleam @@ -6,7 +6,7 @@ import gleeunit/should fn run(id: String, work: turn.Work) -> turn.Run { let pid = process.spawn(fn() { Nil }) - turn.Run(id, pid, process.monitor(pid), False, work) + turn.Run(id, pid, process.monitor(pid), False, turn.latch(), work) } fn running(id: String) -> turn.Activity {