diff --git a/AGENTS.md b/AGENTS.md index e0773b9a..0ed1ff31 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -75,7 +75,8 @@ crates/ ├── tinyhivemind-typesafe/ # exact System One wires and Jev questions behind │ # one transport port; no HTTP client or async runtime ├── tinyhivemind-driver/ # the completion driver over a handle the host binds: -│ # who runs next, and what a committed row means. Pure. +│ # who runs next, what a committed row means, and the +│ # conducted episode: conversations and nudges. Pure. ├── tinyhivemind-openhuman/ # the OpenHuman adapter: both runners behind one seam; │ # the one crate that links a harness, by ADR 0025 ├── tinyhivemind-tools/ # the episode's tools as a record a host drains: diff --git a/crates/tinyhivemind-driver/README.md b/crates/tinyhivemind-driver/README.md index 8b194462..210586f0 100644 --- a/crates/tinyhivemind-driver/README.md +++ b/crates/tinyhivemind-driver/README.md @@ -33,6 +33,14 @@ the current episode participants other than the author. The driver selects each author's fallback from those participants in deterministic scheduling order. +Above the driver sits the `Conductor`: one episode as a host steps it, the +desk and a child episode for every conversation an `ask` opens, with the +rules between them -- conversations run first and conclude to the asker, a +stalled seat or a silent askee is told once, what a wave said lands in the +channel it belongs to, a refused completion is explained, walls end what +will not. It appends nothing: it hands the host notes to append, commits to +append and report the sequence of, and events to log. + The host still owns the runtime and sessions, the transcript, durable append operations, and scheduling. See [`src/README.md`](src/README.md) for the source layout, and `examples/bench/` for the driver priced with no model. diff --git a/crates/tinyhivemind-driver/src/README.md b/crates/tinyhivemind-driver/src/README.md index 576fbfaf..23b4485e 100644 --- a/crates/tinyhivemind-driver/src/README.md +++ b/crates/tinyhivemind-driver/src/README.md @@ -6,4 +6,5 @@ | `error/` | Typed graph, routing, and committed-event failures. | | `graph/` | The owned one-desk graph, `BoundAgent`, and the bindings to its seats. | | `driver/` | Resumable completion rounds, the ledger, the brief, and committed-event folds. | +| `conduct/` | The conductor: the desk episode with its conversations, nudges, sorting, refusals and walls, stepped by a host that appends the rows. | | `test_support.rs` | Test-only seat fixtures shared by unit tests: a name, and an executor. | diff --git a/crates/tinyhivemind-driver/src/conduct/README.md b/crates/tinyhivemind-driver/src/conduct/README.md new file mode 100644 index 00000000..2c82558c --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/README.md @@ -0,0 +1,35 @@ +# `conduct` + +One completion-driven episode as a host steps it: the desk episode, a child +episode for every conversation an `ask` opens, and the rules between them +that no single fold can hold. + +| file | holds | +| --- | --- | +| `mod.rs` | `Conductor`, `ConductPolicy`, `Door`, `starters`; opening the desk, beginning a wave, proposing turns, opening a turn with its brief, recording what it called | +| `wave.rs` | After a wave: the phase machine that hands the host one `Step` at a time -- commits in conversations, silent askees, commits on the desk with their consequences, conclusions, the turn wall | +| `child.rs` | A conversation: its root, its two seats, its own driver state, its turns, its nudge; and one that concluded | +| `steps.rs` | `Turn`, `Note`, `Commit`, `Event`, `Refusal`, `Step` | +| `test.rs` | Every rule, driven by a host that is only a journal | + +The rules, each with the decision it comes from: + +- **A conversation runs first** (ADR 0023): it is what unblocks a desk turn. + It concludes when the seat asked completes, at `child_turn_wall`, or when + nothing is due anywhere; its outcome reaches the asker as a private row, + which releases the asker's hold. The seats that had it are shown it whole + once, on their next desk turn. +- **Nudges** (ADR 0024): a desk seat that holds open work, ran for it and + has been shown everything is told once per assignment and owed a turn. A + seat asked that took its turn without answering is told once and owed a + turn; a second silence stands. +- **Sorting**: a broadcast or an ask made inside a conversation is desk + work; only a post or a completion is a row of the conversation. +- **Refusals**: a completion the ledger refuses is explained to the seat on + the desk. A spent broadcast budget completes the seat with the work. +- **Walls**: turns per conversation, turns per episode. + +The conductor appends nothing. It hands the host a `Note` to append, a +`Commit` to append and report the sequence of, or an `Event` to log, and +takes the sequence back through `committed`. The host owns the journal, +the rendering of a row, the prompt, and running the turn. diff --git a/crates/tinyhivemind-driver/src/conduct/child.rs b/crates/tinyhivemind-driver/src/conduct/child.rs new file mode 100644 index 00000000..5265da5f --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/child.rs @@ -0,0 +1,110 @@ +//! One open conversation: a thread of the desk, run as its own episode. + +use tinyhivemind::Sequence; + +use crate::driver::{ConversationView, DriverState}; + +/// A conversation rooted at an ask row, whose participant is the seat asked, +/// with the asker recorded here (ADR 0023). One question, one answer: the +/// seat asked concludes with `complete_episode`, and its message is the +/// answer; a follow-up is a further ask. +#[derive(Clone, Debug)] +pub(super) struct Child { + pub(super) root: Sequence, + pub(super) asker: String, + pub(super) askee: String, + pub(super) state: DriverState, + /// Turns taken in it so far. + pub(super) turns: u64, + /// The last thing the seat asked said in it: the conclusion, cross-posted. + pub(super) last_by_askee: Option, + /// Whether the seat asked has been told once that it has not answered. + pub(super) nudged: bool, + /// Whether the seat asked took a turn in this wave. + pub(super) turned: bool, +} + +impl Child { + pub(super) fn new(root: Sequence, by: &str, to: &str, state: DriverState) -> Self { + Self { + root, + asker: by.to_owned(), + askee: to.to_owned(), + state, + turns: 0, + last_by_askee: None, + nudged: false, + turned: false, + } + } + + /// Whether `seat` is one of its two. + pub(super) fn involves(&self, seat: &str) -> bool { + self.asker == seat || self.askee == seat + } + + /// The other seat, from `seat`'s side. + pub(super) fn other(&self, seat: &str) -> String { + if seat == self.asker { + self.askee.clone() + } else { + self.asker.clone() + } + } + + /// Over: the seat asked completed, or the wall was reached. + pub(super) fn is_over(&self, wall: u64) -> bool { + self.state.quiescent() || self.turns >= wall + } + + /// What reaches the asker when it concludes. + pub(super) fn outcome(&self, forced: bool) -> String { + if forced { + "the conversation did not conclude in time; take what was said and proceed".to_owned() + } else { + self.last_by_askee + .clone() + .unwrap_or_else(|| "concluded".to_owned()) + } + } + + /// How `seat` sees it, with the transcript the host holds. + pub(super) fn view(&self, seat: &str, transcript: Vec) -> ConversationView { + ConversationView { + root: self.root, + other: self.other(seat), + opened_it: seat == self.asker, + transcript, + concluded: false, + } + } +} + +/// A conversation that concluded, kept for the context of the seats that had +/// it: shown whole once to each, on its next desk turn. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) struct Concluded { + pub(super) root: Sequence, + pub(super) asker: String, + pub(super) askee: String, +} + +impl Concluded { + pub(super) fn involves(&self, seat: &str) -> bool { + self.asker == seat || self.askee == seat + } + + pub(super) fn view(&self, seat: &str, transcript: Vec) -> ConversationView { + ConversationView { + root: self.root, + other: if seat == self.asker { + self.askee.clone() + } else { + self.asker.clone() + }, + opened_it: seat == self.asker, + transcript, + concluded: true, + } + } +} diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs new file mode 100644 index 00000000..deb865fd --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/mod.rs @@ -0,0 +1,451 @@ +//! One completion-driven episode as a host steps it: the desk, its +//! conversations, and the rules between them. +//! +//! [`CompletionDriver`] folds one episode and never runs a turn. A live +//! episode is more than one fold: the desk's, and a child episode for every +//! conversation an `ask` opens (ADR 0023). Between them sit rules the driver +//! cannot hold because they span both, or because they are about what a +//! turn did *not* do: +//! +//! - **Conversations.** An ask roots one at its row, with the seat asked as +//! its participant. It runs ahead of desk turns, because it is what +//! unblocks one. It concludes when the seat asked completes, or at a wall, +//! or when nothing is due anywhere; its outcome is cross-posted to the +//! asker as a private row, which releases the asker's hold. +//! - **Nudges** (ADR 0024). A desk seat that holds open work, ran for it and +//! has been shown everything is told once, per assignment, and owed a turn. +//! A seat asked that took its turn and did not answer is told once and +//! owed a turn; a second silence stands. +//! - **Sorting.** What a wave said lands in the channel it belongs to. A +//! broadcast or an ask made inside a conversation is desk work; anything +//! but a post or a completion inside one is dropped. +//! - **Refusals.** A completion the ledger refuses is explained to the seat +//! on the desk; a spent broadcast budget completes the seat with the work. +//! - **Walls.** Turns per conversation, and turns per episode. +//! +//! [`Conductor`] holds all of that as state and folds. It appends nothing: +//! every row is the host's, so it hands the host [`Step`]s -- a [`Note`] to +//! append, a [`Commit`] to append and report the sequence of, an [`Event`] +//! to log -- and takes the sequence back. One wave, from the host's side: +//! +//! 1. [`begin_wave`](Conductor::begin_wave): the nudges due, as steps. +//! 2. [`turns`](Conductor::turns): who runs, where, and from which row. +//! 3. For each turn, [`open_turn`](Conductor::open_turn) with the rows the +//! host will show it: the [`EpisodeBrief`] for the prompt. +//! 4. Run the turns; [`record`](Conductor::record) what each called. +//! 5. [`step`](Conductor::step) until it returns nothing, appending each +//! note, appending each commit and reporting its sequence through +//! [`committed`](Conductor::committed), and logging each event. +//! +//! [`finished`](Conductor::finished) says when to stop. + +mod child; +mod steps; +#[cfg(test)] +mod test; +mod wave; + +use std::collections::{BTreeMap, BTreeSet}; + +use tinyhivemind::speech::{ToolCall, Utterance}; +use tinyhivemind::{Conversation, Sequence}; +use tinyhivemind_embed::RoutingPlan; +use tinyhivemind_hive::{CompletionEpisodeState, apply_completion}; + +use crate::driver::{BroadcastRouting, Channel, ConversationView, EpisodeBrief}; +use crate::{BoundAgent, CompletionDriver, DriverState, Error, Result}; +use child::{Child, Concluded}; +pub use steps::{Commit, Event, Note, Refusal, Step, Turn}; +use wave::Wave; + +/// The walls a conducted episode runs inside. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ConductPolicy { + /// Turns a conversation may take before it concludes without an answer. + pub child_turn_wall: u64, + /// Turns the whole episode may take before it is abandoned. + pub turn_wall: u64, +} + +impl Default for ConductPolicy { + fn default() -> Self { + Self { + child_turn_wall: 6, + turn_wall: 60, + } + } +} + +/// The door: what the desk is, who sits at it, and who starts. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Door { + /// The chat every tool call names: the desk id. + pub chat: String, + /// The desk's name, for the episode's conversation record. + pub desk_name: String, + /// Every seat, in desk order. + pub members: Vec, + /// The seats the door route starts; the rest are completed at once. + pub starters: Vec, + /// The task's row: where the passed-over seats are completed. + pub opened_at: Sequence, +} + +/// Who the door route starts: the plan's seats, or `fallback` when routing +/// asked for clarification nobody is there to give. +#[must_use] +pub fn starters(plan: &RoutingPlan, fallback: &str) -> Vec { + match plan { + RoutingPlan::One { responder_id, .. } | RoutingPlan::Fallback { responder_id, .. } => { + vec![responder_id.clone()] + } + RoutingPlan::Hive { + primary_id, + invited_ids, + .. + } => std::iter::once(primary_id.clone()) + .chain(invited_ids.iter().cloned()) + .collect(), + RoutingPlan::Clarify { .. } => vec![fallback.to_owned()], + } +} + +/// The desk episode, its conversations, and the rules between them. +pub struct Conductor<'a, A: BoundAgent> { + driver: &'a CompletionDriver<'a, A>, + routing: BroadcastRouting<'a>, + chat: String, + desk_name: String, + policy: ConductPolicy, + state: DriverState, + children: BTreeMap, + concluded: Vec, + /// How many concluded conversations each seat has been shown. + shown: BTreeMap, + /// The assignment each seat was last nudged for on the desk. + desk_nudged: BTreeMap, + turns: u64, + waves: u64, + discharged: u64, + wave: Wave, +} + +impl std::fmt::Debug for Conductor<'_, A> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Conductor") + .field("chat", &self.chat) + .field("turns", &self.turns) + .field("waves", &self.waves) + .field("conversations", &self.children.keys().collect::>()) + .finish_non_exhaustive() + } +} + +impl<'a, A: BoundAgent> Conductor<'a, A> { + /// Open the desk episode. Every member is seated so a handoff can reach + /// any seat; the ones the door route passed over are completed at once + /// at `opened_at`, the task's row -- idle, and reopened by any broadcast + /// that finds them. + /// + /// # Errors + /// + /// [`Error::UnknownStarter`] for a starter that is not a member, and + /// [`Error::NoStarters`] for none at all; otherwise the episode refusing + /// its members, or the driver its start. + pub fn open( + driver: &'a CompletionDriver<'a, A>, + routing: BroadcastRouting<'a>, + policy: ConductPolicy, + door: Door, + ) -> Result { + if door.starters.is_empty() { + return Err(Error::NoStarters); + } + if let Some(seat) = door + .starters + .iter() + .find(|seat| !door.members.contains(seat)) + { + return Err(Error::UnknownStarter { seat: seat.clone() }); + } + let mut episode = CompletionEpisodeState::opened( + Conversation { + desk_id: door.chat.clone(), + desk_name: door.desk_name.clone(), + thread_root: None, + }, + Sequence(0), + door.members.iter().map(String::as_str), + )?; + for id in &door.members { + if !door.starters.contains(id) { + episode = apply_completion(&episode, id, door.opened_at)?; + } + } + let state = driver.start(episode)?; + Ok(Self { + driver, + routing, + chat: door.chat, + desk_name: door.desk_name, + policy, + state, + children: BTreeMap::new(), + concluded: Vec::new(), + shown: BTreeMap::new(), + desk_nudged: BTreeMap::new(), + turns: 0, + waves: 0, + discharged: 0, + wave: Wave::default(), + }) + } + + /// Over: the desk is quiescent and no conversation is open. + #[must_use] + pub fn finished(&self) -> bool { + self.state.quiescent() && self.children.is_empty() + } + + /// The desk episode's state. + #[must_use] + pub const fn state(&self) -> &DriverState { + &self.state + } + + /// The chat every tool call names. + #[must_use] + pub fn chat(&self) -> &str { + &self.chat + } + + /// Seat turns run so far. + #[must_use] + pub const fn turns_run(&self) -> u64 { + self.turns + } + + /// Waves proposed so far. + #[must_use] + pub const fn waves(&self) -> u64 { + self.waves + } + + /// Seats completed with their work for a spent broadcast budget. + #[must_use] + pub const fn discharged(&self) -> u64 { + self.discharged + } + + /// Conversations concluded so far. + #[must_use] + pub fn conversations(&self) -> usize { + self.concluded.len() + } + + /// Start a wave: the desk seats nothing will wake are told once, per + /// assignment, and owed a turn. + pub fn begin_wave(&mut self) -> Vec { + self.waves += 1; + let mut steps = Vec::new(); + for seat in self.state.stalled() { + let assigned_at = self + .state + .episode() + .participants + .iter() + .find(|participant| participant.agent_id == seat) + .and_then(|participant| participant.open()) + .map(|record| record.assigned_at); + if self.desk_nudged.get(&seat) == assigned_at.as_ref() { + continue; + } + if let Some(at) = assigned_at { + self.desk_nudged.insert(seat.clone(), at); + } + steps.push(Step::Event(Event::Nudged { + seat: seat.clone(), + thread: None, + })); + steps.push(Step::Note(Note { + body: "you hold open work and nothing new has arrived. Call `complete_episode` \ + with what you have, or `broadcast` the part that is another seat's. A \ + reply without a tool call records nothing." + .to_owned(), + thread: None, + only_for: Some(seat.clone()), + })); + self.state.owe_turn(&seat); + } + for child in self.children.values_mut() { + child.turned = false; + } + steps + } + + /// The turns due this wave: one per seat, conversations first, because a + /// conversation is what unblocks a desk turn. + /// + /// An empty wave with conversations open concludes them all, without an + /// answer, in the steps that follow. + /// + /// # Errors + /// + /// [`Error::Stalled`]: nothing is due anywhere, no conversation is open, + /// and the desk holds open work. + pub fn turns(&mut self) -> Result> { + let mut taken: BTreeSet = BTreeSet::new(); + let mut turns = Vec::new(); + for child in self.children.values() { + for seat in self.pending(&child.state)? { + if !taken.insert(seat.clone()) { + continue; + } + let since = child + .state + .seen() + .delivered_through + .get(&seat) + .copied() + .unwrap_or(child.root); + turns.push(Turn { + channel: Channel::Thread { + root: child.root, + other: child.other(&seat), + opened_it: seat == child.asker, + }, + seat, + since, + }); + } + } + for seat in self.pending(&self.state)? { + if !taken.insert(seat.clone()) { + continue; + } + let since = self + .state + .seen() + .delivered_through + .get(&seat) + .copied() + .unwrap_or(Sequence(0)); + turns.push(Turn { + seat, + channel: Channel::Desk, + since, + }); + } + if turns.is_empty() && self.children.is_empty() { + return Err(Error::Stalled { + seats: self.state.stalled(), + }); + } + self.wave.begin(turns.is_empty()); + Ok(turns) + } + + fn pending(&self, state: &DriverState) -> Result> { + Ok(self + .driver + .pending_round(state)? + .agents() + .iter() + .map(|pending| pending.hive_agent_id.to_owned()) + .collect()) + } + + /// Open a turn: record that the seat is shown everything through + /// `latest`, and that it ran for what it holds, and build its brief. + /// `new_rows` are the rows above [`Turn::since`] in the turn's channel, + /// rendered by the host; `transcript` is any thread of the desk, whole, + /// for the conversations the seat is or was in. + pub fn open_turn( + &mut self, + turn: &Turn, + latest: Sequence, + new_rows: Vec, + mut transcript: impl FnMut(Sequence) -> Vec, + ) -> EpisodeBrief { + match turn.channel { + Channel::Thread { root, .. } => { + if let Some(child) = self.children.get_mut(&root) { + child.state.delivered(&turn.seat, latest); + child.state.turn_started(&turn.seat); + child.turns += 1; + child.turned = true; + return EpisodeBrief::for_turn( + &child.state, + self.chat.clone(), + turn.seat.clone(), + turn.channel.clone(), + new_rows, + Vec::new(), + ); + } + EpisodeBrief::for_turn( + &self.state, + self.chat.clone(), + turn.seat.clone(), + turn.channel.clone(), + new_rows, + Vec::new(), + ) + } + Channel::Desk => { + self.state.delivered(&turn.seat, latest); + self.state.turn_started(&turn.seat); + let views = self.views(&turn.seat, &mut transcript); + EpisodeBrief::for_turn( + &self.state, + self.chat.clone(), + turn.seat.clone(), + Channel::Desk, + new_rows, + views, + ) + } + } + } + + /// The conversations a seat is shown on a desk turn: those concluded + /// since it last spoke, whole, once; and any still in progress. + fn views( + &mut self, + seat: &str, + transcript: &mut impl FnMut(Sequence) -> Vec, + ) -> Vec { + let cursor = self.shown.entry(seat.to_owned()).or_insert(0); + let mut views: Vec = self.concluded[*cursor..] + .iter() + .filter(|done| done.involves(seat)) + .map(|done| done.view(seat, transcript(done.root))) + .collect(); + *cursor = self.concluded.len(); + views.extend( + self.children + .values() + .filter(|child| child.involves(seat)) + .map(|child| child.view(seat, transcript(child.root))), + ); + views + } + + /// What a turn called, in call order, once the host has closed it. + /// Sorted into the channel each belongs to: a broadcast or an ask made + /// inside a conversation is desk work. + pub fn record(&mut self, turn: &Turn, calls: impl IntoIterator) { + self.turns += 1; + for call in calls { + let ToolCall::Speak(utterance) = call else { + continue; + }; + match (turn.thread(), &utterance) { + (None, _) | (Some(_), Utterance::Broadcast { .. } | Utterance::Ask { .. }) => { + self.wave.desk.push((turn.seat.clone(), utterance)); + } + (Some(root), _) => self.wave.thread.push((root, turn.seat.clone(), utterance)), + } + } + } +} diff --git a/crates/tinyhivemind-driver/src/conduct/steps.rs b/crates/tinyhivemind-driver/src/conduct/steps.rs new file mode 100644 index 00000000..5878b628 --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/steps.rs @@ -0,0 +1,172 @@ +//! What passes between the conductor and its host: a turn to run, and the +//! steps the host takes on the conductor's behalf after a wave. + +use tinyhivemind::Sequence; +use tinyhivemind::speech::Utterance; + +use crate::driver::Channel; + +/// One turn the host runs this wave. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Turn { + /// The seat. + pub seat: String, + /// Where the turn runs: the desk, or a conversation on one of its threads. + pub channel: Channel, + /// The newest row the seat has been shown in this channel: the host + /// gives the turn every row above it. + pub since: Sequence, +} + +impl Turn { + /// The thread the turn runs in, or `None` on the desk. + #[must_use] + pub fn thread(&self) -> Option { + match &self.channel { + Channel::Desk => None, + Channel::Thread { root, .. } => Some(*root), + } + } +} + +/// A row the desk says to a seat: the host appends it, attributed to the +/// desk, and nothing is committed back. The wording is the episode's. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Note { + /// What the desk says. + pub body: String, + /// The thread it lands in, or `None` for the open desk. + pub thread: Option, + /// On the open desk, the one seat it reaches; `None` reaches every seat. + pub only_for: Option, +} + +/// A row the host appends and then commits back with the sequence it got: +/// what a seat said, or what the episode says on a seat's behalf. +/// +/// The host renders the utterance as its own desk row and calls +/// [`Conductor::committed`](super::Conductor::committed) with the sequence. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Commit { + /// The seat the row is attributed to. + pub author: String, + /// What was said. + pub utterance: Utterance, + /// The thread it lands in, or `None` for the open desk. + pub thread: Option, + /// On the open desk, the one seat it reaches; `None` reaches every seat. + pub only_for: Option, + pub(super) kind: Kind, +} + +/// What the conductor does with a commit once it has its sequence. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) enum Kind { + /// A seat spoke in a conversation. + Thread(Sequence), + /// A seat spoke on the desk; a broadcast is routed. + Desk, + /// A conversation concluded: its outcome, cross-posted to the asker. + Conclusion { root: Sequence, forced: bool }, + /// A seat that spent its broadcast budget keeps the work. + Discharge, +} + +/// Why a seat's row was refused, in the terms the desk tells it. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum Refusal { + /// It may not complete: the seats it asked have not answered. + AwaitingReply { + /// The seats it awaits. + waiting_on: Vec, + }, + /// It completed before seeing the assignment it holds. + Undelivered { + /// Where that assignment sits. + assigned_at: Sequence, + }, + /// It spoke in a conversation it has not yet been shown. + NotYetShown, +} + +/// Something the episode did that a host may want to log. Nothing here needs +/// acting on; every consequence is already a [`Note`] or a [`Commit`]. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum Event { + /// A seat was told once that nothing will wake it. + Nudged { + /// The seat. + seat: String, + /// The thread, or `None` on the desk. + thread: Option, + }, + /// A broadcast was placed. + Broadcast { + /// The author. + seat: String, + /// Who took it. + to: Vec, + }, + /// A broadcast fit no seat; the author keeps the work. + Unplaced { + /// The author. + seat: String, + }, + /// A broadcast closed its author's own assignment. + CompletedByBroadcast { + /// The author. + seat: String, + }, + /// An ask opened a conversation. + Asked { + /// The seat that asked. + seat: String, + /// The seat asked. + askee: String, + /// The ask row the conversation is rooted at. + root: Sequence, + }, + /// A queued handoff reached its recipient. + Handoff { + /// The recipient. + to: String, + /// The author of the broadcast it came from. + from: String, + }, + /// A row was refused, and the seat told why. + Refused { + /// The seat. + seat: String, + /// The thread, or `None` on the desk. + thread: Option, + /// Why. + why: Refusal, + }, + /// A seat spent its broadcast budget and was completed with the work. + Discharged { + /// The seat. + seat: String, + }, + /// A conversation concluded. + Concluded { + /// The ask row it was rooted at. + root: Sequence, + /// The seat that asked. + asker: String, + /// The seat asked. + askee: String, + /// Without an answer: nothing was due, or it ran out of turns. + forced: bool, + }, +} + +/// One step the host takes after a wave, in order. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum Step { + /// Append this, attributed to the desk. + Note(Note), + /// Append this and report its sequence. + Commit(Commit), + /// Log this, or don't. + Event(Event), +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs new file mode 100644 index 00000000..ca66792d --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs @@ -0,0 +1,425 @@ +//! Conversations: opened by an ask, run first, concluded to the asker; the silent askee; what is said inside one. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use super::support::{ + Journal, ask, broadcast, complete, door, hive, policy, post, run, seats, two_seat, wave, +}; +use crate::CompletionDriver; +use crate::conduct::{ConductPolicy, Conductor, Event, Refusal, Step}; +use crate::driver::{BroadcastRouting, Channel}; +use tinyhivemind::Sequence; +use tinyhivemind::speech::{ToolCall, Utterance}; + +#[test] +fn an_ask_opens_a_conversation_that_runs_first_and_concludes_to_the_asker() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + + let asked = wave( + &mut conductor, + &journal, + &[("one", vec![ask("two", "what is the port?")])], + ) + .expect("wave"); + let root = Sequence(2); + assert!(matches!( + asked.events.as_slice(), + [Event::Asked { seat, askee, root: at }] if seat == "one" && askee == "two" && *at == root + )); + assert!(!conductor.finished(), "a conversation is open"); + + // The seat asked runs first, in the thread; the asker, held, does not run + // on the desk. It answers by completing. + let answered = wave( + &mut conductor, + &journal, + &[("two", vec![complete("port 8080")])], + ) + .expect("wave"); + assert_eq!( + seats(&answered.turns)[0], + ("two", Some(root)), + "the askee in the thread runs first; the asker, woken by its own ask row, after" + ); + assert_eq!(answered.turns[0].since, root); + assert!(matches!( + answered.turns[0].channel, + Channel::Thread { root: at, ref other, opened_it: false } if at == root && other == "one" + )); + assert!(matches!( + answered.events.as_slice(), + [Event::Concluded { root: at, asker, askee, forced: false }] + if *at == root && asker == "one" && askee == "two" + )); + assert_eq!(conductor.conversations(), 1); + let to_asker = journal.private_to("one"); + assert_eq!( + to_asker, + vec!["concluded our conversation (thread 2): port 8080"], + "the answer reaches the asker as a private row" + ); + + // The asker is released: it runs on the desk, is shown the whole + // conversation once, and completes. + let turns = conductor.turns().expect("turns"); + let brief = conductor.open_turn(&turns[0], journal.latest(), Vec::new(), |root| { + journal.thread(root) + }); + assert_eq!(brief.conversations.len(), 1); + assert!(brief.conversations[0].concluded); + assert!(brief.conversations[0].opened_it); + assert_eq!(brief.conversations[0].other, "two"); + assert_eq!( + brief.conversations[0].transcript.len(), + 2, + "the ask and the answer" + ); + conductor.record(&turns[0], vec![ToolCall::Speak(complete("shipped"))]); + while let Some(step) = conductor.step().expect("steps") { + if let Step::Commit(commit) = step { + let sequence = journal.append(&commit.author, "row", commit.thread, None); + run(conductor.committed(sequence)).expect("committed"); + } + } + assert!(conductor.finished()); + // Shown once: a second desk turn would show nothing again. + let brief = conductor.open_turn(&turns[0], journal.latest(), Vec::new(), |_| Vec::new()); + assert!(brief.conversations.is_empty()); +} + +#[test] +fn a_completion_while_a_conversation_is_open_is_refused_and_explained() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + // The asker asks and completes in the same turn: the ask opens the + // conversation, and the completion is refused because of it. + let seen = wave( + &mut conductor, + &journal, + &[("one", vec![ask("two", "?"), complete("too soon")])], + ) + .expect("wave"); + assert!(seen.events.iter().any(|event| matches!( + event, + Event::Refused { seat, thread: None, why: Refusal::AwaitingReply { waiting_on } } + if seat == "one" && waiting_on == &["two".to_owned()] + ))); + assert!( + journal + .private_to("one") + .iter() + .any(|body| body.contains("your completion was refused")), + "{:?}", + journal.bodies() + ); +} + +#[test] +fn a_silent_askee_is_nudged_once_and_then_walled() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat( + &driver, + routing, + ConductPolicy { + child_turn_wall: 3, + turn_wall: 60, + }, + &journal, + ); + wave(&mut conductor, &journal, &[("one", vec![ask("two", "?")])]).expect("wave"); + // Turn one in the thread: a post, no answer. Told once, owed a turn. + let posted = wave(&mut conductor, &journal, &[("two", vec![post("thinking")])]).expect("wave"); + assert!(matches!( + posted.events.as_slice(), + [Event::Nudged { seat, thread: Some(_) }] if seat == "two" + )); + assert!( + journal + .thread(Sequence(2)) + .iter() + .any(|row| row.contains("is waiting")) + ); + // Turn two: silence. No second nudge; still owed nothing, so the thread + // runs it once more because the nudge owed it a turn. + let silent = wave(&mut conductor, &journal, &[]).expect("wave"); + assert_eq!(seats(&silent.turns)[0], ("two", Some(Sequence(2)))); + assert!( + !silent.events.iter().any(|event| matches!( + event, + Event::Nudged { + thread: Some(_), + .. + } + )), + "a second silence stands: {:?}", + silent.events + ); + // Turn three reaches the wall: concluded without an answer, forced. + let walled = wave( + &mut conductor, + &journal, + &[("two", vec![post("still thinking")])], + ) + .expect("wave"); + assert!( + walled + .events + .iter() + .any(|event| matches!(event, Event::Concluded { forced: true, .. })), + "{:?}", + walled.events + ); + assert!( + journal + .private_to("one") + .iter() + .any(|body| body.contains("did not conclude in time")) + ); +} + +#[test] +fn a_conversation_at_its_wall_concludes_without_a_nudge() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat( + &driver, + routing, + ConductPolicy { + child_turn_wall: 1, + turn_wall: 60, + }, + &journal, + ); + wave(&mut conductor, &journal, &[("one", vec![ask("two", "?")])]).expect("wave"); + // The first thread turn is the last: the wall is one. It concludes, and + // the askee is not told to answer a conversation that is already over. + let walled = wave(&mut conductor, &journal, &[("two", vec![post("hm")])]).expect("wave"); + assert!( + walled + .events + .iter() + .any(|event| matches!(event, Event::Concluded { forced: true, .. })), + "{:?}", + walled.events + ); + assert!( + !walled.events.iter().any(|event| matches!( + event, + Event::Nudged { + thread: Some(_), + .. + } + )), + "{:?}", + walled.events + ); + assert!( + !journal + .thread(Sequence(2)) + .iter() + .any(|row| row.contains("is waiting")), + "{:?}", + journal.thread(Sequence(2)) + ); +} + +#[test] +fn a_refused_reply_in_a_conversation_is_not_its_answer() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + wave(&mut conductor, &journal, &[("one", vec![ask("two", "?")])]).expect("wave"); + // The askee completes in the thread before being shown its assignment: + // the host opened its turn at a watermark below the ask row, so the fold + // refuses the row. + conductor.begin_wave(); + let turns = conductor.turns().expect("turns"); + let thread_turn = turns + .iter() + .find(|turn| turn.seat == "two") + .expect("the askee is due"); + conductor.open_turn(thread_turn, Sequence(1), Vec::new(), |_| Vec::new()); + conductor.record(thread_turn, vec![ToolCall::Speak(complete("too early"))]); + let mut refused = false; + while let Some(step) = conductor.step().expect("steps") { + match step { + Step::Commit(commit) => { + let sequence = journal.append(&commit.author, "row", commit.thread, None); + run(conductor.committed(sequence)).expect("committed"); + } + Step::Event(Event::Refused { + why: Refusal::NotYetShown, + .. + }) => refused = true, + Step::Event(_) | Step::Note(_) => {} + } + } + assert!(refused, "the early completion was refused"); + // The conversation goes on to conclude without an answer: the refused + // message was never the conversation's. + let mut concluded = false; + for _ in 0..8 { + let seen = wave(&mut conductor, &journal, &[]).expect("wave"); + if seen + .events + .iter() + .any(|event| matches!(event, Event::Concluded { .. })) + { + concluded = true; + break; + } + } + assert!(concluded, "the conversation concluded"); + assert!( + journal + .private_to("one") + .iter() + .all(|body| !body.contains("too early")), + "{:?}", + journal.private_to("one") + ); +} + +#[test] +fn a_broadcast_or_ask_inside_a_conversation_is_desk_work_and_a_dm_is_dropped() { + let hive = hive(&["one", "two", "three"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two", "three"], &["one"], &journal), + ) + .expect("opens"); + wave(&mut conductor, &journal, &[("one", vec![ask("two", "?")])]).expect("wave"); + // Inside the thread, two broadcasts to the desk, dms nobody, and answers. + let seen = wave( + &mut conductor, + &journal, + &[( + "two", + vec![ + broadcast("three should check the logs"), + Utterance::Dm { + to: vec!["one".into()], + message: "psst".into(), + }, + complete("answered"), + ], + )], + ) + .expect("wave"); + assert!( + seen.events + .iter() + .any(|event| matches!(event, Event::Broadcast { seat, to } if seat == "two" && !to.is_empty())), + "{:?}", + seen.events + ); + assert!( + seen.events + .iter() + .any(|event| matches!(event, Event::Concluded { .. })) + ); + assert!( + !journal.bodies().iter().any(|body| body == "psst"), + "a dm in a thread is not served, so it is not a row" + ); + // The broadcast landed on the desk as desk work for a third seat. + let turns = conductor.turns().expect("turns"); + assert!( + seats(&turns) + .iter() + .any(|(seat, thread)| *seat == "three" && thread.is_none()) + ); +} + +#[test] +fn nothing_due_concludes_every_open_conversation_without_an_answer() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + wave(&mut conductor, &journal, &[("one", vec![ask("two", "?")])]).expect("wave"); + // The askee is nudged after its first silent turn, runs once more, and + // then nothing is due: the conversation is forced closed. + wave(&mut conductor, &journal, &[]).expect("wave"); + wave(&mut conductor, &journal, &[]).expect("wave"); + let forced = wave(&mut conductor, &journal, &[]).expect("wave"); + assert!(forced.turns.is_empty()); + assert!( + forced + .events + .iter() + .any(|event| matches!(event, Event::Concluded { forced: true, .. })), + "{:?}", + forced.events + ); + assert_eq!(conductor.conversations(), 1); +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/desk.rs b/crates/tinyhivemind-driver/src/conduct/test/desk.rs new file mode 100644 index 00000000..7672876a --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/desk.rs @@ -0,0 +1,249 @@ +//! The desk: stalled seats, broadcasts placed and unplaced, handoffs, the budget, the walls, and the commit protocol. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use super::support::{ + ClarifyRouter, Journal, broadcast, complete, door, hive, policy, post, run, seats, two_seat, + wave, +}; +use crate::conduct::{ConductPolicy, Conductor, Event, Refusal, Step}; +use crate::driver::BroadcastRouting; +use crate::{CompletionDriver, Error}; +use tinyhivemind::Sequence; +use tinyhivemind::speech::ToolCall; + +#[test] +fn a_stalled_desk_seat_is_nudged_once_per_assignment_and_then_the_episode_stalls() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + // One runs and calls nothing. Next wave it is stalled: nudged, owed a turn. + wave(&mut conductor, &journal, &[]).expect("wave"); + let nudged = wave(&mut conductor, &journal, &[]).expect("wave"); + assert!(matches!( + nudged.events.as_slice(), + [Event::Nudged { seat, thread: None }] if seat == "one" + )); + assert_eq!(seats(&nudged.turns), vec![("one", None)]); + assert!(journal.private_to("one")[0].contains("you hold open work")); + // Silent again, for the same assignment: no second nudge, nothing due. + let stalled = wave(&mut conductor, &journal, &[]); + assert!( + matches!(&stalled, Err(Error::Stalled { seats }) if seats == &["one".to_owned()]), + "{stalled:?}" + ); +} + +#[test] +fn an_unplaced_broadcast_leaves_the_work_with_the_author_and_says_so() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: Some(&ClarifyRouter), + reasoning: Some(&ClarifyRouter), + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + let seen = wave( + &mut conductor, + &journal, + &[("one", vec![broadcast("someone take this")])], + ) + .expect("wave"); + assert!(matches!( + seen.events.as_slice(), + [Event::Unplaced { seat }] if seat == "one" + )); + assert!(journal.private_to("one")[0].contains("nobody on this desk can take that")); + assert!(!conductor.finished(), "the author keeps the work"); +} + +#[test] +fn a_placed_broadcast_completes_its_author_and_a_busy_recipient_gets_it_as_a_handoff() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4) + .expect("driver") + .with_queue_depth(2) + .expect("depth"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + // Both start: two holds work, so one's broadcast to it is queued. + let mut conductor = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two"], &["one", "two"], &journal), + ) + .expect("opens"); + let seen = wave( + &mut conductor, + &journal, + &[("one", vec![broadcast("two: also check the cache")])], + ) + .expect("wave"); + assert!( + seen.events.iter().any(|event| matches!( + event, + Event::CompletedByBroadcast { seat } if seat == "one" + )), + "{:?}", + seen.events + ); + // Two completes its own work and is handed the queued broadcast. + let handed = wave( + &mut conductor, + &journal, + &[("two", vec![complete("mine is done")])], + ) + .expect("wave"); + assert!( + handed.events.iter().any(|event| matches!( + event, + Event::Handoff { to, from } if to == "two" && from == "one" + )), + "{:?}", + handed.events + ); + assert!(journal.private_to("two")[0].contains("handoff from @one")); + assert!(!conductor.finished()); + // A completion before it is shown the handoff is refused and explained; + // the host shows nothing new by opening the turn at a stale watermark. + let turns = conductor.turns().expect("turns"); + assert_eq!(seats(&turns), vec![("two", None)]); + conductor.record(&turns[0], vec![ToolCall::Speak(complete("that too"))]); + let mut refused = None; + while let Some(step) = conductor.step().expect("steps") { + match step { + Step::Commit(commit) => { + let sequence = journal.append(&commit.author, "row", None, None); + run(conductor.committed(sequence)).expect("committed"); + } + Step::Event(Event::Refused { why, .. }) => refused = Some(why), + Step::Event(_) | Step::Note(_) => {} + } + } + assert!( + matches!(refused, Some(Refusal::Undelivered { .. })), + "{refused:?}" + ); +} + +#[test] +fn a_spent_broadcast_budget_completes_the_seat_with_the_work() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4) + .expect("driver") + .with_broadcast_budget(Some(1)); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: Some(&ClarifyRouter), + reasoning: Some(&ClarifyRouter), + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + // Two unplaced broadcasts from one assignment: the second is over budget. + let seen = wave( + &mut conductor, + &journal, + &[("one", vec![broadcast("first"), broadcast("second")])], + ) + .expect("wave"); + assert!( + seen.events + .iter() + .any(|event| matches!(event, Event::Discharged { seat } if seat == "one")), + "{:?}", + seen.events + ); + assert_eq!(conductor.discharged(), 1); + assert!( + journal + .bodies() + .iter() + .any(|body| body == "COMPLETE: budget spent; keeping the work") + ); + assert!(conductor.finished()); +} + +#[test] +fn the_turn_wall_ends_the_episode() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat( + &driver, + routing, + ConductPolicy { + child_turn_wall: 6, + turn_wall: 1, + }, + &journal, + ); + let walled = wave(&mut conductor, &journal, &[("one", vec![post("hm")])]); + assert!( + matches!(walled, Err(Error::TurnWall { wall: 1 })), + "{walled:?}" + ); +} + +#[test] +fn a_commit_must_be_reported_before_the_next_step_and_only_once() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + assert!(matches!( + run(conductor.committed(Sequence(9))), + Err(Error::NoCommitOutstanding) + )); + conductor.begin_wave(); + let turns = conductor.turns().expect("turns"); + conductor.record(&turns[0], vec![ToolCall::Speak(post("a row"))]); + let step = conductor.step().expect("step").expect("a commit"); + assert!( + matches!(&step, Step::Commit(commit) if commit.author == "one" && commit.thread.is_none()) + ); + assert!(matches!(conductor.step(), Err(Error::CommitOutstanding))); + run(conductor.committed(Sequence(2))).expect("committed"); + assert!(conductor.step().expect("settles").is_none()); + assert_eq!(ConductPolicy::default().child_turn_wall, 6); +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/door.rs b/crates/tinyhivemind-driver/src/conduct/test/door.rs new file mode 100644 index 00000000..5abacf0c --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/door.rs @@ -0,0 +1,117 @@ +//! The door: who starts, and every shape a plan takes. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use super::support::{ClarifyRouter, Journal, complete, door, hive, policy, run, seats, wave}; +use crate::conduct::{ConductPolicy, Conductor, starters}; +use crate::driver::BroadcastRouting; +use crate::{CompletionDriver, Error}; +use tinyhivemind_embed::{Router, RoutingFallback, RoutingPlan, RoutingRequest}; + +#[test] +fn the_door_starts_the_routed_seats_and_completes_the_rest() { + let hive = hive(&["one", "two", "three"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two", "three"], &["one"], &journal), + ) + .expect("opens"); + assert!(!conductor.finished()); + assert_eq!(conductor.chat(), "engineering"); + assert!(format!("{conductor:?}").contains("engineering")); + + let first = wave(&mut conductor, &journal, &[("one", vec![complete("done")])]).expect("wave"); + assert_eq!( + seats(&first.turns), + vec![("one", None)], + "only the starter runs" + ); + assert!(conductor.finished(), "one seat's completion ends it"); + assert_eq!(conductor.turns_run(), 1); + assert_eq!(conductor.waves(), 1); + assert_eq!(conductor.discharged(), 0); + assert_eq!(conductor.conversations(), 0); + assert_eq!(journal.bodies(), vec!["the task", "COMPLETE: done"]); + assert!(conductor.state().quiescent()); +} + +#[test] +fn the_door_refuses_a_starter_outside_the_desk_and_no_starter_at_all() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let outsider = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two"], &["three"], &journal), + ); + assert!( + matches!(outsider, Err(Error::UnknownStarter { ref seat }) if seat == "three"), + "{outsider:?}" + ); + let nobody = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two"], &[], &journal), + ); + assert!(matches!(nobody, Err(Error::NoStarters)), "{nobody:?}"); +} + +#[test] +fn starters_are_read_from_every_plan_shape() { + let evaluation = run(ClarifyRouter.evaluate(&RoutingRequest { + message: String::new(), + source: tinyhivemind_embed::RoutingSource::DeskMessage, + conversation: tinyhivemind_embed::ConversationRef { + id: "engineering".into(), + kind: tinyhivemind_embed::ConversationKind::Desk, + thread_root: None, + }, + desk_purpose: None, + thread_context: Vec::new(), + candidates: hive(&["a", "b"]).graph().candidates.clone(), + roster_version: 1, + policy: policy(1), + })) + .expect("evaluates"); + let one = RoutingPlan::One { + responder_id: "a".into(), + evaluation: evaluation.clone(), + }; + let fallback = RoutingPlan::Fallback { + responder_id: "b".into(), + reason: RoutingFallback::ProviderUnavailable, + }; + let hive = RoutingPlan::Hive { + primary_id: "c".into(), + invited_ids: vec!["d".into(), "e".into()], + evaluation: evaluation.clone(), + }; + let clarify = RoutingPlan::Clarify { evaluation }; + assert_eq!(starters(&one, "z"), vec!["a"]); + assert_eq!(starters(&fallback, "z"), vec!["b"]); + assert_eq!(starters(&hive, "z"), vec!["c", "d", "e"]); + assert_eq!(starters(&clarify, "z"), vec!["z"]); +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/mod.rs b/crates/tinyhivemind-driver/src/conduct/test/mod.rs new file mode 100644 index 00000000..3ce1eba6 --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/mod.rs @@ -0,0 +1,7 @@ +//! The conductor: conversations, nudges, sorting, refusals and walls, driven +//! by a host that is only a journal. + +mod conversations; +mod desk; +mod door; +mod support; diff --git a/crates/tinyhivemind-driver/src/conduct/test/support.rs b/crates/tinyhivemind-driver/src/conduct/test/support.rs new file mode 100644 index 00000000..c8aa92ce --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/support.rs @@ -0,0 +1,309 @@ +//! Shared fixtures: a hive of named seats, a router that always asks for +//! clarification, a journal, and one wave driven the way a host drives it. + +#![allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)] +use std::sync::Mutex; + +use tinyhivemind::desk::{Desk, ResponderMode}; +use tinyhivemind::speech::{ToolCall, Utterance}; +use tinyhivemind::{Sequence, responder::Probability}; +use tinyhivemind_embed::{ + CandidateProbability, ContributionProbability, EvaluationDisposition, RouteCandidate, Router, + RouterFuture, RoutingEvaluation, RoutingPolicy, RoutingRequest, +}; + +use crate::conduct::{ConductPolicy, Conductor, Door, Event, Step, Turn}; +use crate::driver::BroadcastRouting; +use crate::test_support::Seat; +use crate::{AgentBinding, BoundHive, CompletionDriver, Error, HiveGraph}; + +pub(super) fn run(future: F) -> F::Output { + tokio::runtime::Builder::new_current_thread() + .build() + .expect("runtime") + .block_on(future) +} + +pub(super) fn probability(parts: u32) -> Probability { + Probability::new(parts).expect("bounded") +} + +pub(super) fn policy(round_width: usize) -> RoutingPolicy { + RoutingPolicy { + minimum_confidence: probability(350_000), + high_impact_minimum_confidence: probability(800_000), + clarification_threshold: probability(850_000), + high_impact_threshold: probability(700_000), + round_width, + choice_option_limit: 8, + } +} + +pub(super) fn hive(ids: &[&str]) -> BoundHive { + BoundHive::new( + HiveGraph::new( + Desk { + id: "engineering".into(), + name: "Engineering".into(), + description: None, + members: ids.iter().map(|id| (*id).into()).collect(), + responder_mode: ResponderMode::Auto, + }, + ids.iter() + .map(|id| RouteCandidate { + id: (*id).into(), + label: (*id).into(), + role: None, + description: None, + capabilities: Vec::new(), + learned_topics: Vec::new(), + available: true, + }) + .collect(), + ), + ids.iter() + .map(|id| AgentBinding::new(*id, Seat((*id).to_owned()))) + .collect(), + ) + .expect("hive") +} + +/// An evaluation that asks for clarification on every pass, so a broadcast +/// is placed with nobody. +#[derive(Debug)] +pub(super) struct ClarifyRouter; + +impl Router for ClarifyRouter { + fn evaluate<'a>(&'a self, request: &'a RoutingRequest) -> RouterFuture<'a> { + let eligible: Vec = request + .candidates + .iter() + .map(|candidate| candidate.id.clone()) + .collect(); + let roster_version = request.roster_version; + Box::pin(async move { + if eligible.is_empty() { + return Err("no candidates to route among".into()); + } + let share = 1_000_000 / (u32::try_from(eligible.len()).expect("small") + 1); + let mut primary_probabilities: Vec = eligible + .iter() + .map(|id| CandidateProbability { + candidate_id: id.clone(), + probability: probability(share), + }) + .collect(); + primary_probabilities.push(CandidateProbability { + candidate_id: "none".into(), + probability: probability( + 1_000_000 - share * u32::try_from(eligible.len()).expect("small"), + ), + }); + Ok(RoutingEvaluation { + primary_responder: eligible[0].clone(), + primary_probabilities, + confidence: probability(900_000), + needs_collaboration: probability(0), + needs_clarification: probability(1_000_000), + contributions: eligible + .iter() + .map(|id| ContributionProbability { + candidate_id: id.clone(), + probability: probability(500_000), + }) + .collect(), + high_impact: probability(0), + model_identity: "test".into(), + question_schema_version: 1, + roster_version, + disposition: EvaluationDisposition::Unchecked, + }) + }) + } +} + +/// One row: sequence, author, body, thread, and the one seat it is for. +pub(super) type Row = (Sequence, String, String, Option, Option); + +/// The host: rows, and nothing else. +#[derive(Debug, Default)] +pub(super) struct Journal { + rows: Mutex>, +} + +impl Journal { + pub(super) fn append( + &self, + author: &str, + body: &str, + thread: Option, + only_for: Option, + ) -> Sequence { + let mut rows = self.rows.lock().unwrap(); + let sequence = Sequence(rows.last().map_or(0, |row| row.0.0) + 1); + rows.push((sequence, author.into(), body.into(), thread, only_for)); + sequence + } + + pub(super) fn latest(&self) -> Sequence { + self.rows + .lock() + .unwrap() + .last() + .map_or(Sequence(0), |row| row.0) + } + + pub(super) fn thread(&self, root: Sequence) -> Vec { + self.rows + .lock() + .unwrap() + .iter() + .filter(|row| row.0 == root || row.3 == Some(root)) + .map(|row| format!("@{}: {}", row.1, row.2)) + .collect() + } + + pub(super) fn bodies(&self) -> Vec { + self.rows + .lock() + .unwrap() + .iter() + .map(|row| row.2.clone()) + .collect() + } + + pub(super) fn private_to(&self, seat: &str) -> Vec { + self.rows + .lock() + .unwrap() + .iter() + .filter(|row| row.4.as_deref() == Some(seat)) + .map(|row| row.2.clone()) + .collect() + } +} + +/// What one wave produced, as the host saw it. +#[derive(Debug, Default)] +pub(super) struct Wave { + pub(super) turns: Vec, + pub(super) events: Vec, +} + +/// One wave: nudges, turns, the scripted calls each seat makes, and every +/// step after, appended to the journal. +pub(super) fn wave( + conductor: &mut Conductor<'_, Seat>, + journal: &Journal, + calls: &[(&str, Vec)], +) -> Result { + let mut seen = Wave::default(); + for step in conductor.begin_wave() { + take(step, journal, &mut seen); + } + let turns = conductor.turns()?; + for turn in &turns { + let brief = conductor.open_turn(turn, journal.latest(), Vec::new(), |root| { + journal.thread(root) + }); + assert_eq!(brief.seat, turn.seat); + let script = calls + .iter() + .find(|(seat, _)| *seat == turn.seat) + .map(|(_, calls)| calls.clone()) + .unwrap_or_default(); + conductor.record(turn, script.into_iter().map(ToolCall::Speak)); + } + seen.turns = turns; + while let Some(step) = conductor.step()? { + if let Step::Commit(commit) = &step { + let sequence = journal.append( + &commit.author, + &describe(&commit.utterance), + commit.thread, + commit.only_for.clone(), + ); + run(conductor.committed(sequence))?; + continue; + } + take(step, journal, &mut seen); + } + Ok(seen) +} + +pub(super) fn take(step: Step, journal: &Journal, seen: &mut Wave) { + match step { + Step::Note(note) => { + journal.append("desk", ¬e.body, note.thread, note.only_for); + } + Step::Event(event) => seen.events.push(event), + Step::Commit(_) => panic!("a commit is not taken, it is committed"), + } +} + +pub(super) fn describe(utterance: &Utterance) -> String { + match utterance { + Utterance::Post { message } | Utterance::Dm { message, .. } => message.clone(), + Utterance::Broadcast { message } => format!("BROADCAST: {message}"), + Utterance::Ask { to, message } => format!("asks @{to}: {message}"), + Utterance::CompleteEpisode { message } => format!("COMPLETE: {message}"), + } +} + +pub(super) fn complete(message: &str) -> Utterance { + Utterance::CompleteEpisode { + message: message.into(), + } +} + +pub(super) fn ask(to: &str, message: &str) -> Utterance { + Utterance::Ask { + to: to.into(), + message: message.into(), + } +} + +pub(super) fn broadcast(message: &str) -> Utterance { + Utterance::Broadcast { + message: message.into(), + } +} + +pub(super) fn post(message: &str) -> Utterance { + Utterance::Post { + message: message.into(), + } +} + +pub(super) fn door(ids: &[&str], starters: &[&str], journal: &Journal) -> Door { + let opened_at = journal.append("operator", "the task", None, None); + Door { + chat: "engineering".into(), + desk_name: "Engineering".into(), + members: ids.iter().map(|id| (*id).into()).collect(), + starters: starters.iter().map(|id| (*id).into()).collect(), + opened_at, + } +} + +pub(super) fn seats(turns: &[Turn]) -> Vec<(&str, Option)> { + turns + .iter() + .map(|turn| (turn.seat.as_str(), turn.thread())) + .collect() +} + +pub(super) fn two_seat<'a>( + driver: &'a CompletionDriver<'a, Seat>, + routing: BroadcastRouting<'a>, + policy: ConductPolicy, + journal: &Journal, +) -> Conductor<'a, Seat> { + Conductor::open( + driver, + routing, + policy, + door(&["one", "two"], &["one"], journal), + ) + .expect("opens") +} diff --git a/crates/tinyhivemind-driver/src/conduct/wave.rs b/crates/tinyhivemind-driver/src/conduct/wave.rs new file mode 100644 index 00000000..79ec69f2 --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/wave.rs @@ -0,0 +1,482 @@ +//! After a wave: what the seats said, committed in order, and what follows. +//! +//! The host appends every row, so this is a phase machine the host steps: +//! [`Conductor::step`] hands out one [`Step`] at a time, and a [`Commit`] +//! is not followed by another step until the host has reported its +//! sequence through [`Conductor::committed`]. The phases, in order: what +//! was said in conversations; the seats asked that said nothing; what was +//! said on the desk, with its consequences; the conversations that are +//! over; the turn wall. + +use std::collections::VecDeque; + +use tinyhivemind::Sequence; +use tinyhivemind::speech::Utterance; +use tinyhivemind_hive::CompletionEpisodeState; + +use super::Conductor; +use super::child::{Child, Concluded}; +use super::steps::{Commit, Event, Kind, Note, Refusal, Step}; +use crate::driver::{BroadcastRouting, CommittedUtterance, HostAction, Transition}; +use tinyhivemind::Conversation; + +use crate::{BoundAgent, Error, Result}; + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +enum Phase { + /// No wave in progress. + #[default] + Idle, + /// Commit what was said in conversations. + Threads, + /// Tell the seats asked that said nothing. + SilentAskees, + /// Commit what was said on the desk. + Desk, + /// Conclude the conversations that are over. + Conclude, + /// Check the turn wall. + Wall, +} + +/// One wave's bookkeeping. +#[derive(Debug, Default)] +pub(super) struct Wave { + phase: Phase, + /// Nothing was due: every open conversation concludes without an answer. + force_conclusions: bool, + /// What thread turns said: `(root, seat, utterance)`. + pub(super) thread: Vec<(Sequence, String, Utterance)>, + /// What desk turns said, and what thread turns said to the desk. + pub(super) desk: Vec<(String, Utterance)>, + /// Steps ready for the host, notes and events. + steps: VecDeque, + /// Commits waiting for the host, in order. + commits: VecDeque, + /// The commit the host holds and has not reported. + outstanding: Option, +} + +impl Wave { + pub(super) fn begin(&mut self, nothing_due: bool) { + self.phase = Phase::Threads; + self.force_conclusions = nothing_due; + } + + fn event(&mut self, event: Event) { + self.steps.push_back(Step::Event(event)); + } + + fn note(&mut self, body: impl Into, thread: Option, only_for: Option<&str>) { + self.steps.push_back(Step::Note(Note { + body: body.into(), + thread, + only_for: only_for.map(str::to_owned), + })); + } +} + +impl<'a, A: BoundAgent> Conductor<'a, A> { + /// The next step after a wave, or `None` when the wave is settled. + /// + /// # Errors + /// + /// [`Error::CommitOutstanding`] when the last commit's sequence has not + /// been reported; [`Error::TurnWall`] when the episode has run past it. + pub fn step(&mut self) -> Result> { + loop { + if let Some(step) = self.wave.steps.pop_front() { + return Ok(Some(step)); + } + if self.wave.outstanding.is_some() { + return Err(Error::CommitOutstanding); + } + if let Some(commit) = self.wave.commits.pop_front() { + self.wave.outstanding = Some(commit.clone()); + return Ok(Some(Step::Commit(commit))); + } + match self.wave.phase { + Phase::Idle => return Ok(None), + Phase::Threads => { + for (root, seat, utterance) in std::mem::take(&mut self.wave.thread) { + self.queue_thread(root, seat, utterance); + } + self.wave.phase = Phase::SilentAskees; + } + Phase::SilentAskees => { + // Nothing due means every conversation concludes now; a + // nudge would owe a turn nobody will run. + if !self.wave.force_conclusions { + self.nudge_silent_askees(); + } + self.wave.phase = Phase::Desk; + } + Phase::Desk => { + for (seat, utterance) in std::mem::take(&mut self.wave.desk) { + let only_for = utterance.asks().map(str::to_owned); + self.wave.commits.push_back(Commit { + author: seat, + utterance, + thread: None, + only_for, + kind: Kind::Desk, + }); + } + self.wave.phase = Phase::Conclude; + } + Phase::Conclude => { + self.queue_conclusions(); + self.wave.phase = Phase::Wall; + } + Phase::Wall => { + self.wave.phase = Phase::Idle; + if self.turns >= self.policy.turn_wall { + return Err(Error::TurnWall { + wall: self.policy.turn_wall, + }); + } + } + } + } + } + + /// A row committed to a conversation. Only a post or a completion can be + /// said inside one; `dm` is not served, and the rest went to the desk. + fn queue_thread(&mut self, root: Sequence, seat: String, utterance: Utterance) { + if !self.children.contains_key(&root) + || !matches!( + utterance, + Utterance::Post { .. } | Utterance::CompleteEpisode { .. } + ) + { + return; + } + self.wave.commits.push_back(Commit { + author: seat, + utterance, + thread: Some(root), + only_for: None, + kind: Kind::Thread(root), + }); + } + + /// The seat asked took its turn and the conversation is not over: it did + /// not answer, whatever it did instead. Once, it is told so and owed one + /// more turn; a second silence stands. A conversation at its wall is + /// concluding this wave and is not nudged. + fn nudge_silent_askees(&mut self) { + let wall = self.policy.child_turn_wall; + for child in self.children.values_mut() { + if child.turned && !child.is_over(wall) && !child.nudged { + child.nudged = true; + self.wave.steps.push_back(Step::Note(Note { + body: "the seat that asked you is waiting: answer with `complete_episode`, \ + and its message is your answer. If you need another seat first, say \ + so in that answer." + .to_owned(), + thread: Some(child.root), + only_for: None, + })); + child.state.owe_turn(&child.askee); + self.wave.steps.push_back(Step::Event(Event::Nudged { + seat: child.askee.clone(), + thread: Some(child.root), + })); + } + } + } + + /// Conversations that ended this wave, or ran past their wall, or were + /// left with nothing due anywhere, conclude: their outcome is + /// cross-posted to the asker, which releases its hold. + fn queue_conclusions(&mut self) { + let wall = self.policy.child_turn_wall; + let force = self.wave.force_conclusions; + let over: Vec = self + .children + .iter() + .filter(|(_, child)| force || child.is_over(wall)) + .map(|(root, _)| *root) + .collect(); + for root in over { + let Some(child) = self.children.get(&root) else { + continue; + }; + let forced = !child.state.quiescent(); + self.wave.commits.push_back(Commit { + author: child.askee.clone(), + utterance: Utterance::Dm { + to: vec![child.asker.clone()], + message: format!( + "concluded our conversation (thread {}): {}", + root.0, + child.outcome(forced) + ), + }, + thread: None, + only_for: Some(child.asker.clone()), + kind: Kind::Conclusion { root, forced }, + }); + } + } + + /// The sequence the host gave the outstanding commit. + /// + /// # Errors + /// + /// [`Error::NoCommitOutstanding`] when nothing was handed out, or any + /// driver error the fold could not explain to the seat. + pub async fn committed(&mut self, sequence: Sequence) -> Result<()> { + let commit = self + .wave + .outstanding + .take() + .ok_or(Error::NoCommitOutstanding)?; + let committed = CommittedUtterance { + author_id: commit.author.clone(), + sequence, + utterance: commit.utterance.clone(), + }; + match commit.kind { + Kind::Thread(root) => self.commit_thread(root, committed).await, + Kind::Desk => self.commit_desk(committed).await, + Kind::Conclusion { root, forced } => { + self.commit_conclusion(root, forced, committed).await + } + Kind::Discharge => { + let transition = self + .driver + .apply_committed(&self.state, committed, None) + .await?; + self.state = transition.state; + Ok(()) + } + } + } + + async fn commit_thread(&mut self, root: Sequence, committed: CommittedUtterance) -> Result<()> { + let Some(child) = self.children.get_mut(&root) else { + return Ok(()); + }; + let seat = committed.author_id.clone(); + let said = committed.utterance.message().to_owned(); + match self + .driver + .apply_committed(&child.state, committed, None) + .await + { + Ok(transition) => { + child.state = transition.state; + // The answer is what the fold accepted, not what was tried. + if seat == child.askee { + child.last_by_askee = Some(said); + } + } + Err(Error::UndeliveredAssignment { .. }) => self.wave.event(Event::Refused { + seat, + thread: Some(root), + why: Refusal::NotYetShown, + }), + Err(error) => return Err(error), + } + Ok(()) + } + + fn routing(&self) -> BroadcastRouting<'a> { + self.routing + } + + async fn commit_desk(&mut self, committed: CommittedUtterance) -> Result<()> { + let seat = committed.author_id.clone(); + let sequence = committed.sequence; + let asked = committed.utterance.asks().map(str::to_owned); + let is_broadcast = committed.utterance.broadcasting(); + let held = holds(&self.state, &seat); + let routing = self.routing(); + match self + .driver + .apply_committed(&self.state, committed, Some(routing)) + .await + { + Ok(transition) => { + let said = Said { + seat: seat.clone(), + sequence, + asked, + is_broadcast, + held, + }; + self.consequences(&said, transition)?; + } + Err(Error::AwaitingReply { waiting_on, .. }) => { + self.wave.note( + format!( + "your completion was refused: your conversation with {} has not \ + concluded. Its outcome reaches you on a later turn; complete after it \ + does.", + waiting_on.join(", ") + ), + None, + Some(&seat), + ); + self.wave.event(Event::Refused { + seat, + thread: None, + why: Refusal::AwaitingReply { waiting_on }, + }); + } + Err(Error::UndeliveredAssignment { assigned_at, .. }) => { + self.wave.note( + format!( + "you were handed new work at sequence {} while you were speaking; it is \ + in your next messages. Your completion applied to nothing.", + assigned_at.0 + ), + None, + Some(&seat), + ); + self.wave.event(Event::Refused { + seat, + thread: None, + why: Refusal::Undelivered { assigned_at }, + }); + } + Err(Error::BudgetSpent { .. }) => { + self.discharged += 1; + self.wave.event(Event::Discharged { seat: seat.clone() }); + // Before whatever else the wave said: the seat keeps the work + // now, so a later row from it applies to that. + self.wave.commits.push_front(Commit { + author: seat, + utterance: Utterance::CompleteEpisode { + message: "budget spent; keeping the work".into(), + }, + thread: None, + only_for: None, + kind: Kind::Discharge, + }); + } + Err(error) => return Err(error), + } + Ok(()) + } + + /// What follows from a desk row the fold accepted: broadcasts placed or + /// not, a conversation opened, handoffs delivered. + fn consequences(&mut self, said: &Said, transition: Transition) -> Result<()> { + let mut routed = false; + for action in transition.actions { + match action { + HostAction::RunAgents { agent_ids, .. } => { + routed = true; + self.wave.event(Event::Broadcast { + seat: said.seat.clone(), + to: agent_ids, + }); + } + // For an ask, this is the signal to open the conversation: a + // thread of the desk rooted at the ask row, with the two as + // its seats. + HostAction::DeliverDm { .. } => { + if let Some(askee) = &said.asked { + self.open_conversation(&said.seat, askee, said.sequence)?; + } + } + HostAction::DeliverHandoff { agent_id, handoff } => { + self.wave.event(Event::Handoff { + to: agent_id.clone(), + from: handoff.from.clone(), + }); + self.wave.note( + format!("handoff from @{}: {}", handoff.from, handoff.body), + None, + Some(&agent_id), + ); + } + } + } + if said.is_broadcast && said.held && !holds(&transition.state, &said.seat) { + self.wave.event(Event::CompletedByBroadcast { + seat: said.seat.clone(), + }); + } + if said.is_broadcast && !routed { + self.wave.event(Event::Unplaced { + seat: said.seat.clone(), + }); + self.wave.note( + "nobody on this desk can take that; the work stays with you. Do what you can \ + with what the desk holds, or complete with what you have.", + None, + Some(&said.seat), + ); + } + self.state = transition.state; + Ok(()) + } + + fn open_conversation(&mut self, by: &str, to: &str, root: Sequence) -> Result<()> { + let state = self.driver.start(CompletionEpisodeState::opened( + Conversation { + desk_id: self.chat.clone(), + desk_name: self.desk_name.clone(), + thread_root: Some(root), + }, + root, + [to], + )?)?; + self.wave.event(Event::Asked { + seat: by.to_owned(), + askee: to.to_owned(), + root, + }); + self.children.insert(root, Child::new(root, by, to, state)); + Ok(()) + } + + async fn commit_conclusion( + &mut self, + root: Sequence, + forced: bool, + committed: CommittedUtterance, + ) -> Result<()> { + let Some(child) = self.children.remove(&root) else { + return Ok(()); + }; + let transition = self + .driver + .apply_committed(&self.state, committed, None) + .await?; + self.state = transition.state; + self.wave.event(Event::Concluded { + root, + asker: child.asker.clone(), + askee: child.askee.clone(), + forced, + }); + self.concluded.push(Concluded { + root, + asker: child.asker, + askee: child.askee, + }); + Ok(()) + } +} + +/// A desk row as it was, before the fold moved anything. +struct Said { + seat: String, + sequence: Sequence, + asked: Option, + is_broadcast: bool, + held: bool, +} + +/// Whether `seat` holds an open assignment on the desk. +fn holds(state: &crate::DriverState, seat: &str) -> bool { + state + .episode() + .participants + .iter() + .any(|participant| participant.agent_id == seat && participant.open().is_some()) +} diff --git a/crates/tinyhivemind-driver/src/error/mod.rs b/crates/tinyhivemind-driver/src/error/mod.rs index 62290ba0..33c644db 100644 --- a/crates/tinyhivemind-driver/src/error/mod.rs +++ b/crates/tinyhivemind-driver/src/error/mod.rs @@ -241,6 +241,35 @@ pub enum Error { /// The underlying completion fold rejected the event. #[error(transparent)] Completion(#[from] tinyhivemind_hive::Error), + /// Nothing is due anywhere, no conversation is open, and the desk holds + /// open work: nobody is owed a turn, so nothing will move. + #[error("episode stalled with {seats:?} holding open work")] + Stalled { + /// The seats holding it. + seats: Vec, + }, + /// The episode ran past its turn wall. + #[error("turn wall of {wall} reached")] + TurnWall { + /// The wall. + wall: u64, + }, + /// The host asked for the next step while holding a commit it has not + /// reported the sequence of. + #[error("a commit is outstanding; report its sequence first")] + CommitOutstanding, + /// The host reported a sequence for a commit it was never handed. + #[error("no commit is outstanding")] + NoCommitOutstanding, + /// The door named a starter that is not a member of the desk. + #[error("starter `{seat}` is not a member of the desk")] + UnknownStarter { + /// The seat. + seat: String, + }, + /// The door named nobody to start. + #[error("the door names no starter")] + NoStarters, } /// The crate-wide result alias. diff --git a/crates/tinyhivemind-driver/src/lib.rs b/crates/tinyhivemind-driver/src/lib.rs index 2ef08af1..ebbe9ffb 100644 --- a/crates/tinyhivemind-driver/src/lib.rs +++ b/crates/tinyhivemind-driver/src/lib.rs @@ -69,6 +69,7 @@ //! # } //! ``` +pub mod conduct; pub mod driver; pub mod error; pub mod graph; @@ -76,6 +77,9 @@ pub mod graph; #[cfg(test)] mod test_support; +pub use conduct::{ + Commit, ConductPolicy, Conductor, Door, Event, Note, Refusal, Step, Turn, starters, +}; pub use driver::{ AssignmentSpend, BroadcastRouting, Channel, CommittedUtterance, CompletionDriver, ConversationView, DriverState, EpisodeBrief, Handoff, HostAction, Ledger, PendingAgent, diff --git a/crates/tinyhivemind-openhuman/src/lib.rs b/crates/tinyhivemind-openhuman/src/lib.rs index 9f815c7b..4d38aabc 100644 --- a/crates/tinyhivemind-openhuman/src/lib.rs +++ b/crates/tinyhivemind-openhuman/src/lib.rs @@ -31,12 +31,14 @@ //! //! # Example //! -//! A raw seat, offline, against the scripted model the `offline` feature -//! ships. The same steps seat a live one: the route is the credential. +//! A raw seat, against any OpenAI-compatible endpoint: the route is the +//! credential. The same steps seat one against the scripted model the +//! `offline` feature ships, which is how the crate's own tests prove it. //! //! ```no_run //! use std::sync::Arc; -//! use tinyhivemind_openhuman::{RawRunner, Route, SeatRunner, offline}; +//! use openhuman_embed::RuntimeConfig; +//! use tinyhivemind_openhuman::{Lane, RawRunner, Route, SeatRunner}; //! use tinyhivemind_tools::{Dispatch, EpisodeTools}; //! //! # async fn run() -> tinyhivemind_openhuman::Result<()> { @@ -48,14 +50,18 @@ //! Arc::new(EpisodeTools::new(["lead"])), //! &[("lead".to_owned(), "You lead the desk.".to_owned())].into_iter().collect(), //! "Call `complete_episode` when you are done.", -//! &offline::config(), +//! &RuntimeConfig::default(), //! "http://127.0.0.1:1/backend", -//! &Route { endpoint: "http://127.0.0.1:1/v1".into(), api_key: "key".into(), model: offline::MODEL.into() }, +//! &Route { +//! endpoint: "http://127.0.0.1:1/v1".into(), +//! api_key: "key".into(), +//! model: "a-model".into(), +//! }, //! &workspace, //! ) //! .await?; //! runner.open("lead", Vec::new(), Dispatch { chat: "engineering".into(), parent: None }); -//! let (_, _, reply) = runner.turn("lead".into(), tinyhivemind_openhuman::Lane::Desk, "Go.".into()).await; +//! let (_, _, reply) = runner.turn("lead".into(), Lane::Desk, "Go.".into()).await; //! let events = runner.close("lead"); //! # let _ = (reply, events); //! # Ok(()) diff --git a/examples/openhuman/README.md b/examples/openhuman/README.md index 8015427f..05df044e 100644 --- a/examples/openhuman/README.md +++ b/examples/openhuman/README.md @@ -58,8 +58,11 @@ corpus and paid campaign described in ## `conducted`: one loop, two runners `src/bin/conducted.rs` steps one completion-driven episode the way a host steps -it: propose a round, run it, commit what it said, report delivery, repeat until -quiescent. The journal, the lanes, the briefs and the driver are the host's. +it: begin a wave, run the turns the `Conductor` proposes, record what each +called, then append the notes and commits it hands back until the wave +settles. The journal, the prompt and the log are the host's; the +conversations, nudges, sorting, refusals and walls are the conductor's, in +`tinyhivemind-driver`. How a seat's turn *runs* is behind one seam, `SeatRunner`, with two implementations the loop cannot tell apart: diff --git a/examples/openhuman/src/bin/conducted.rs b/examples/openhuman/src/bin/conducted.rs index e1dccfeb..b3482f0d 100644 --- a/examples/openhuman/src/bin/conducted.rs +++ b/examples/openhuman/src/bin/conducted.rs @@ -35,16 +35,15 @@ use conducted::jev::LiveJev; use openhuman_embed::{Access, Provider, Runtime, RuntimeConfig, Workspace}; use tinyhivemind::desk::{Desk, ResponderMode}; use tinyhivemind::responder::Probability; -use tinyhivemind::speech::{ToolCall, Utterance}; -use tinyhivemind::{Conversation, Sequence}; +use tinyhivemind::speech::Utterance; +use tinyhivemind::Sequence; use tinyhivemind_embed::{ ConversationKind, ConversationRef, RouteCandidate, Router, RouterFuture, RoutingPlan, RoutingPolicy, RoutingRequest, RoutingSource, route_message, }; -use tinyhivemind_hive::{CompletionEpisodeState, apply_completion}; use tinyhivemind_driver::{ - BoundAgent, BoundHive, BroadcastRouting, Channel, CommittedUtterance, CompletionDriver, - ConversationView, DriverState, EpisodeBrief, Error, HiveGraph, HostAction, standing_contract, + BoundHive, BroadcastRouting, CompletionDriver, ConductPolicy, Conductor, Door, Event, + HiveGraph, Refusal, Step, standing_contract, }; use tinyhivemind_openhuman::{ EmbedRunner, Lane, RawRunner, Route, RunnerKind, SeatRunner, TurnJob, offline, @@ -196,10 +195,6 @@ const REPLY_SHOWN: usize = 600; const JEV_MODEL: &str = "jev-1.13.0"; const OPENROUTER: &str = "https://openrouter.ai/api/v1"; const WORKER_STACK_BYTES: usize = 16 * 1024 * 1024; -/// Hard wall on turns: a chain that will not end is a finding, not a hang. -const TURN_WALL: u64 = 60; -/// Turns one conversation may take before it is concluded without an answer. -const CHILD_TURN_WALL: u64 = 6; fn main() -> anyhow::Result<()> { tokio::runtime::Builder::new_multi_thread() @@ -293,24 +288,6 @@ fn render(row: &Row) -> String { format!("@{}: {}", row.author, row.body) } -/// One open conversation: a thread of the desk, run as its own episode whose -/// participant is the seat asked, with the asker recorded here (ADR 0023). One -/// question, one answer: the seat asked concludes with `complete_episode`, and -/// its message is the answer; a follow-up is a further ask. -struct Child { - root: Sequence, - asker: String, - askee: String, - state: DriverState, - turns: u64, - /// The last thing the seat asked said in it: the conclusion, cross-posted. - last_by_askee: Option, - /// Whether the seat asked has been told once that it has not answered. - nudged: bool, - /// Whether the seat asked took a turn in this wave. - turned: bool, -} - /// A router that counts its calls: the provider bill, one line. struct Counted { inner: R, @@ -686,7 +663,10 @@ struct Report { /// One episode over `runner`, stepped to quiescence. /// /// Generic rather than boxed so each runner's bound seat type flows into the -/// hive and the driver as itself: the loop never names it. +/// hive and the driver as itself: the loop never names it. What is here is +/// only what a host owns: the journal, the prompt, running a turn, and the +/// log. The rules -- conversations, nudges, what a wave said and where it +/// goes, refusals, walls -- are the conductor's. async fn episode(runner: R, setup: Setup) -> anyhow::Result { let Setup { scenario, @@ -734,11 +714,9 @@ async fn episode(runner: R, setup: Setup) -> anyhow::Result = router.as_ref().map(|r| r as &(dyn Router + '_)); let journal = Journal::default(); - journal.append("operator", scenario.task, None, None); + let opened_at = journal.append("operator", scenario.task, None, None); - // The door route: who starts. Everyone is seated so a handoff can reach - // any seat; the ones the route passed over are completed at once -- idle, - // and reopened by any broadcast that finds them. + // The door route: who starts. let door = RoutingRequest { message: scenario.task.to_owned(), source: RoutingSource::DeskMessage, @@ -754,38 +732,11 @@ async fn episode(runner: R, setup: Setup) -> anyhow::Result { - vec![responder_id.clone()] - } - RoutingPlan::Hive { - primary_id, - invited_ids, - .. - } => std::iter::once(primary_id.clone()) - .chain(invited_ids.iter().cloned()) - .collect(), - RoutingPlan::Clarify { .. } => { - eprintln!("[door] routing asked for clarification; lead owns it"); - vec![fallback.to_owned()] - } - }; - println!("[door] starts: {}", starters.join(", ")); - - let mut episode = CompletionEpisodeState::opened( - Conversation { - desk_id: desk_id.into(), - desk_name: scenario.name.into(), - thread_root: None, - }, - Sequence(0), - ids.iter().map(String::as_str), - )?; - for id in &ids { - if !starters.contains(id) { - episode = apply_completion(&episode, id, Sequence(1))?; - } + if matches!(plan, RoutingPlan::Clarify { .. }) { + eprintln!("[door] routing asked for clarification; {fallback} owns it"); } + let starters = tinyhivemind_driver::starters(&plan, fallback); + println!("[door] starts: {}", starters.join(", ")); // Round width four: the door can start up to four seats and they run // together. The broadcast policy is width one -- a handoff belongs to one @@ -801,218 +752,61 @@ async fn episode(runner: R, setup: Setup) -> anyhow::Result = BTreeMap::new(); - // Concluded conversations, kept whole for the context of the seats that - // had them: `(root, asker, askee, transcript)`, and how many each seat has - // already been shown. - let mut concluded: Vec<(Sequence, String, String, Vec)> = Vec::new(); - let mut shown: BTreeMap = BTreeMap::new(); - let mut desk_nudged: BTreeMap = BTreeMap::new(); - let mut turns = 0_u64; - let mut waves = 0_u64; - let mut discharged = 0_u64; - let settled = 'episode: loop { - if state.quiescent() && children.is_empty() { + let settled: anyhow::Result<()> = 'episode: loop { + if conductor.finished() { break Ok(()); } - waves += 1; - - // A seat that holds open desk work, ran for it, and has been shown - // everything is stalled: nothing will wake it. Once per assignment it - // is told, and owed one more turn. - for seat_id in state.stalled() { - let assigned_at = state - .episode() - .participants - .iter() - .find(|participant| participant.agent_id == seat_id) - .and_then(|participant| participant.open()) - .map(|record| record.assigned_at); - if desk_nudged.get(&seat_id) == assigned_at.as_ref() { - continue; - } - if let Some(at) = assigned_at { - desk_nudged.insert(seat_id.clone(), at); - } - eprintln!("[nudged] @{seat_id} on the desk: stalled with open work"); - journal.append( - "desk", - "you hold open work and nothing new has arrived. Call `complete_episode` \ - with what you have, or `broadcast` the part that is another seat's. A reply \ - without a tool call records nothing.", - None, - Some(&seat_id), - ); - state.owe_turn(&seat_id); - } - for child in children.values_mut() { - child.turned = false; + for step in conductor.begin_wave() { + take(&journal, step); } - - // One turn per seat per wave, and conversations first: a conversation - // is what unblocks a desk turn, so it goes ahead of it. - let mut taken: BTreeSet = BTreeSet::new(); + let turns = match conductor.turns() { + Ok(turns) => turns, + Err(error) => break Err(error.into()), + }; let mut jobs: Vec = Vec::new(); - for child in children.values_mut() { - // Collected first: the round borrows the state it was proposed - // from, and preparing a turn reports delivery into that state. - let seats: Vec = driver - .pending_round(&child.state)? - .agents() - .iter() - .map(|pending| pending.hive_agent_id.to_owned()) - .collect(); - for seat_id in seats { - if !taken.insert(seat_id.clone()) { - continue; - } - let through = child - .state - .seen() - .delivered_through - .get(&seat_id) - .copied() - .unwrap_or(child.root); - let rows = journal.thread_since(child.root, through); - runner.open( - &seat_id, - journal.thread(child.root), - Dispatch { - chat: desk_id.into(), - parent: Some(child.root.0.to_string()), - }, - ); - child.state.delivered(&seat_id, journal.latest()); - child.state.turn_started(&seat_id); - child.turns += 1; - child.turned = true; - let other = if seat_id == child.asker { - child.askee.clone() - } else { - child.asker.clone() - }; - let brief = EpisodeBrief::for_turn( - &child.state, - desk_id, - &seat_id, - Channel::Thread { - root: child.root, - other: other.clone(), - opened_it: seat_id == child.asker, - }, - rows, - Vec::new(), - ); - let prompt = format!( - "## The desk\n{DESK_PREAMBLE}\n\n## Who you are\n{}\n\n{}", - briefs[&seat_id], - brief.render() - ); - jobs.push(runner.turn(seat_id, Lane::Thread(child.root), prompt)); - } - } - let seats: Vec = driver - .pending_round(&state)? - .agents() - .iter() - .map(|pending| pending.hive_agent_id.to_owned()) - .collect(); - for seat_id in seats { - if !taken.insert(seat_id.clone()) { - continue; - } - let through = state - .seen() - .delivered_through - .get(&seat_id) - .copied() - .unwrap_or(Sequence(0)); - let rows = journal.desk_since(&seat_id, through); + for turn in &turns { + let rows = match turn.thread() { + None => journal.desk_since(&turn.seat, turn.since), + Some(root) => journal.thread_since(root, turn.since), + }; runner.open( - &seat_id, - rows.clone(), + &turn.seat, + match turn.thread() { + None => rows.clone(), + Some(root) => journal.thread(root), + }, Dispatch { chat: desk_id.into(), - parent: None, + parent: turn.thread().map(|root| root.0.to_string()), }, ); - state.delivered(&seat_id, journal.latest()); - state.turn_started(&seat_id); - // The conversations this seat had: concluded since it last spoke, - // shown whole once, and any still in progress -- its shared - // context across every channel it is in. - let cursor = shown.entry(seat_id.clone()).or_insert(0); - let mut views: Vec = concluded[*cursor..] - .iter() - .filter(|(_, asker, askee, _)| *asker == seat_id || *askee == seat_id) - .map(|(root, asker, askee, transcript)| ConversationView { - root: *root, - other: if *asker == seat_id { - askee.clone() - } else { - asker.clone() - }, - opened_it: *asker == seat_id, - transcript: transcript.clone(), - concluded: true, - }) - .collect(); - *cursor = concluded.len(); - views.extend( - children - .values() - .filter(|child| child.asker == seat_id || child.askee == seat_id) - .map(|child| ConversationView { - root: child.root, - other: if child.asker == seat_id { - child.askee.clone() - } else { - child.asker.clone() - }, - opened_it: child.asker == seat_id, - transcript: journal.thread(child.root), - concluded: false, - }), - ); - let brief = - EpisodeBrief::for_turn(&state, desk_id, &seat_id, Channel::Desk, rows, views); + let brief = conductor.open_turn(turn, journal.latest(), rows, |root| { + journal.thread(root) + }); // What the host owns first; what the episode knows after. let prompt = format!( "## The desk\n{DESK_PREAMBLE}\n\n## Who you are\n{}\n\n{}", - briefs[&seat_id], + briefs[&turn.seat], brief.render() ); - jobs.push(runner.turn(seat_id, Lane::Desk, prompt)); - } - if jobs.is_empty() { - // Nothing is due anywhere. A conversation nobody will continue is - // concluded without an answer; a desk with open work and nobody - // owed a turn is stalled. - let stuck: Vec = children.keys().copied().collect(); - if stuck.is_empty() { - break Err(anyhow::anyhow!( - "episode stalled with {:?} holding open work", - state.stalled() - )); - } - for root in stuck { - let child = children.remove(&root).expect("listed"); - state = conclude(&driver, &journal, &state, child, true, &mut concluded).await?; - } - continue; + let lane = turn.thread().map_or(Lane::Desk, Lane::Thread); + jobs.push(runner.turn(turn.seat.clone(), lane, prompt)); } let outcomes = futures::future::join_all(jobs).await; - - // Everything the wave said, sorted into the channel it belongs to. - // A broadcast made inside a conversation is desk work: the seat found - // something for someone else while talking, and refusing it there - // cost a live run both of its deliverables. - let mut desk_events: Vec<(String, Utterance)> = Vec::new(); - let mut thread_events: Vec<(Sequence, String, Utterance)> = Vec::new(); for (seat_id, lane, outcome) in outcomes { - turns += 1; let where_ = match lane { Lane::Desk => String::new(), Lane::Thread(root) => format!(" in thread {}", root.0), @@ -1048,240 +842,42 @@ async fn episode(runner: R, setup: Setup) -> anyhow::Result { - desk_events.push((seat_id.clone(), utterance)); - } - (Lane::Thread(root), _) => { - thread_events.push((root, seat_id.clone(), utterance)) - } - } - } - } - - for (root, seat_id, utterance) in thread_events { - let Some(child) = children.get_mut(&root) else { - continue; - }; - if !matches!( - utterance, - Utterance::Post { .. } | Utterance::CompleteEpisode { .. } - ) { - // Only `dm` can reach here, and it is not served. - continue; - } - let sequence = journal.append(&seat_id, &describe(&utterance), Some(root), None); - if seat_id == child.askee { - child.last_by_askee = Some(utterance.message().to_owned()); - } - let committed = CommittedUtterance { - author_id: seat_id.clone(), - sequence, - utterance, - }; - match driver.apply_committed(&child.state, committed, None).await { - Ok(transition) => child.state = transition.state, - Err(Error::UndeliveredAssignment { .. }) => { - eprintln!("[refused] @{seat_id} in thread {}: not yet shown", root.0); - } - Err(error) => break 'episode Err(error.into()), - } - } - - // The seat asked took its turn and the conversation is not over: it - // did not answer, whatever it did instead. Once, it is told so and - // owed one more turn; a second silence stands. - for child in children.values_mut() { - if child.turned && !child.state.quiescent() && !child.nudged { - child.nudged = true; - journal.append( - "desk", - "the seat that asked you is waiting: answer with `complete_episode`, and its \ - message is your answer. If you need another seat first, say so in that answer.", - Some(child.root), - None, - ); - child.state.owe_turn(&child.askee); - eprintln!("[nudged] @{} in thread {}", child.askee, child.root.0); - } + let turn = turns + .iter() + .find(|turn| turn.seat == seat_id) + .expect("every outcome is a turn that was proposed"); + conductor.record(turn, events.into_iter().map(|event| event.call)); } - - for (seat_id, utterance) in desk_events { - // An ask is private to the seat it asks and roots a conversation; - // everything else is the desk's. - let asked = utterance.asks().map(str::to_owned); - let sequence = journal.append(&seat_id, &describe(&utterance), None, asked.as_deref()); - let is_broadcast = utterance.broadcasting(); - let held = holds(&state, &seat_id); - let committed = CommittedUtterance { - author_id: seat_id.clone(), - sequence, - utterance, - }; - match driver - .apply_committed(&state, committed, Some(routing)) - .await - { - Ok(transition) => { - let mut routed = false; - for action in &transition.actions { - match action { - HostAction::RunAgents { agent_ids, .. } => { - routed = true; - println!("[broadcast] @{seat_id} -> {}", agent_ids.join(", ")); - } - // For an ask, this is the signal to open the - // conversation: a thread of the desk rooted at the - // ask row, with the two as its seats. - HostAction::DeliverDm { .. } => { - if let Some(askee) = &asked { - let child_state = - driver.start(CompletionEpisodeState::opened( - Conversation { - desk_id: desk_id.into(), - desk_name: scenario.name.into(), - thread_root: Some(sequence), - }, - sequence, - [askee.as_str()], - )?)?; - println!( - "[ask] @{seat_id} opened a conversation with @{askee} (thread {})", - sequence.0 - ); - children.insert( - sequence, - Child { - root: sequence, - asker: seat_id.clone(), - askee: askee.clone(), - state: child_state, - turns: 0, - last_by_askee: None, - nudged: false, - turned: false, - }, - ); - } - } - HostAction::DeliverHandoff { agent_id, handoff } => { - println!( - "[handoff] -> @{agent_id} (queued from @{})", - handoff.from - ); - journal.append( - "desk", - &format!("handoff from @{}: {}", handoff.from, handoff.body), - None, - Some(agent_id), - ); - } - } - } - if is_broadcast && held && !holds(&transition.state, &seat_id) { - eprintln!("[completed] @{seat_id} by its broadcast"); - } - if is_broadcast && !routed { - println!( - "[unplaced] @{seat_id}'s broadcast fits no seat; it keeps the work" - ); - journal.append( - "desk", - "nobody on this desk can take that; the work stays with you. Do what \ - you can with what the desk holds, or complete with what you have.", - None, - Some(&seat_id), - ); - } - state = transition.state; - } - Err(Error::AwaitingReply { waiting_on, .. }) => { - eprintln!( - "[refused] @{seat_id} may not complete: in conversation with {waiting_on:?}" - ); - journal.append( - "desk", - &format!( - "your completion was refused: your conversation with {} has not \ - concluded. Its outcome reaches you on a later turn; complete after it \ - does.", - waiting_on.join(", ") - ), - None, - Some(&seat_id), + loop { + match conductor.step() { + Ok(None) => break, + Ok(Some(Step::Commit(commit))) => { + let sequence = journal.append( + &commit.author, + &describe(&commit.utterance), + commit.thread, + commit.only_for.as_deref(), ); + if let Err(error) = conductor.committed(sequence).await { + break 'episode Err(error.into()); + } } - Err(Error::UndeliveredAssignment { assigned_at, .. }) => { - eprintln!( - "[refused] @{seat_id} completed before seeing its assignment at {}", - assigned_at.0 - ); - journal.append( - "desk", - &format!( - "you were handed new work at sequence {} while you were speaking; it \ - is in your next messages. Your completion applied to nothing.", - assigned_at.0 - ), - None, - Some(&seat_id), - ); - } - Err(Error::BudgetSpent { .. }) => { - eprintln!( - "[refused] @{seat_id} has spent its broadcast budget; it keeps the work" - ); - discharged += 1; - let sequence = - journal.append(&seat_id, "budget spent; keeping the work", None, None); - let transition = driver - .apply_committed( - &state, - CommittedUtterance { - author_id: seat_id.clone(), - sequence, - utterance: Utterance::CompleteEpisode { - message: "budget spent; keeping the work".into(), - }, - }, - None, - ) - .await?; - state = transition.state; - } + Ok(Some(step)) => take(&journal, step), Err(error) => break 'episode Err(error.into()), } } - - // Conversations that ended this wave, or ran past their wall, conclude: - // their outcome is cross-posted to the asker, which releases its hold. - let over: Vec = children - .iter() - .filter(|(_, child)| child.state.quiescent() || child.turns >= CHILD_TURN_WALL) - .map(|(root, _)| *root) - .collect(); - for root in over { - let child = children.remove(&root).expect("listed"); - let forced = !child.state.quiescent(); - state = conclude(&driver, &journal, &state, child, forced, &mut concluded).await?; - } - if turns >= TURN_WALL { - break Err(anyhow::anyhow!("turn wall of {TURN_WALL} reached")); - } }; println!( - "turns {turns} | routes {} | waves {waves} | discharged {discharged} | settled {} | conversations {}", + "turns {} | routes {} | waves {} | discharged {} | settled {} | conversations {}", + conductor.turns_run(), router .as_ref() .map_or(0, |r| r.calls.load(Ordering::SeqCst)), - state.episode().settled(), - concluded.len() + conductor.waves(), + conductor.discharged(), + conductor.state().episode().settled(), + conductor.conversations() ); for row in journal.all().iter().filter(|_| !quiet) { let scope = match (row.thread, row.only_for.as_deref()) { @@ -1314,73 +910,79 @@ async fn episode(runner: R, setup: Setup) -> anyhow::Result( - driver: &CompletionDriver<'_, A>, - journal: &Journal, - state: &DriverState, - child: Child, - forced: bool, - concluded: &mut Vec<(Sequence, String, String, Vec)>, -) -> anyhow::Result { - let transcript = journal.thread(child.root); - let outcome = if forced { - "the conversation did not conclude in time; take what was said and proceed".to_owned() - } else { - child - .last_by_askee - .clone() - .unwrap_or_else(|| "concluded".to_owned()) - }; - println!( - "[concluded] thread {} between @{} and @{}{}", - child.root.0, - child.asker, - child.askee, - if forced { - " (nothing due, or out of turns)" - } else { - "" +/// A note the desk says, appended; an event, logged. +fn take(journal: &Journal, step: Step) { + match step { + Step::Note(note) => { + journal.append("desk", ¬e.body, note.thread, note.only_for.as_deref()); } - ); - let sequence = journal.append( - &child.askee, - &format!( - "concluded our conversation (thread {}): {outcome}", - child.root.0 + Step::Event(event) => log(&event), + Step::Commit(_) => unreachable!("a commit is appended and reported, not taken"), + } +} + +/// The log line for what the episode did. +fn log(event: &Event) { + match event { + Event::Nudged { seat, thread: None } => { + eprintln!("[nudged] @{seat} on the desk: stalled with open work"); + } + Event::Nudged { + seat, + thread: Some(root), + } => eprintln!("[nudged] @{seat} in thread {}", root.0), + Event::Broadcast { seat, to } => println!("[broadcast] @{seat} -> {}", to.join(", ")), + Event::Unplaced { seat } => { + println!("[unplaced] @{seat}'s broadcast fits no seat; it keeps the work"); + } + Event::CompletedByBroadcast { seat } => eprintln!("[completed] @{seat} by its broadcast"), + Event::Asked { seat, askee, root } => println!( + "[ask] @{seat} opened a conversation with @{askee} (thread {})", + root.0 ), - None, - Some(&child.asker), - ); - let transition = driver - .apply_committed( - state, - CommittedUtterance { - author_id: child.askee.clone(), - sequence, - utterance: Utterance::Dm { - to: vec![child.asker.clone()], - message: outcome, - }, - }, - None, - ) - .await?; - concluded.push((child.root, child.asker, child.askee, transcript)); - Ok(transition.state) + Event::Handoff { to, from } => println!("[handoff] -> @{to} (queued from @{from})"), + Event::Refused { seat, thread, why } => { + let where_ = thread.map_or(String::new(), |root| format!(" in thread {}", root.0)); + let reason = match why { + Refusal::AwaitingReply { waiting_on } => { + format!("may not complete: in conversation with {waiting_on:?}") + } + Refusal::Undelivered { assigned_at } => format!( + "completed before seeing its assignment at {}", + assigned_at.0 + ), + Refusal::NotYetShown => "not yet shown".to_owned(), + }; + eprintln!("[refused] @{seat}{where_}: {reason}"); + } + Event::Discharged { seat } => { + eprintln!("[refused] @{seat} has spent its broadcast budget; it keeps the work"); + } + Event::Concluded { + root, + asker, + askee, + forced, + } => println!( + "[concluded] thread {} between @{asker} and @{askee}{}", + root.0, + if *forced { + " (nothing due, or out of turns)" + } else { + "" + } + ), + } } /// How a row reads on the desk. @@ -1399,15 +1001,6 @@ fn required(name: &str) -> anyhow::Result { std::env::var(name).map_err(|_| anyhow::anyhow!("{name} must be set for a live run")) } -/// Whether `seat` holds an open assignment on the desk. -fn holds(state: &DriverState, seat: &str) -> bool { - state - .episode() - .participants - .iter() - .any(|participant| participant.agent_id == seat && participant.open().is_some()) -} - fn candidate(id: &str, role: &str) -> RouteCandidate { RouteCandidate { id: id.to_owned(),