diff --git a/Cargo.lock b/Cargo.lock index 78dd47f8..5f1113a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2844,6 +2844,7 @@ dependencies = [ "thiserror", "tinyhivemind", "tinyhivemind-driver", + "tinyhivemind-embed", "tinyhivemind-mcp", "tinyhivemind-tools", "tinytools", diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs index 39331a7a..5e649064 100644 --- a/crates/tinyhivemind-driver/src/conduct/mod.rs +++ b/crates/tinyhivemind-driver/src/conduct/mod.rs @@ -50,7 +50,7 @@ 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 tinyhivemind_hive::CompletionEpisodeState; use crate::driver::{BroadcastRouting, Channel, ConversationView, EpisodeBrief}; use crate::{BoundAgent, CompletionDriver, DriverState, Error, Result}; @@ -168,18 +168,24 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { { return Err(Error::UnknownStarter { seat: seat.clone() }); } + // Everyone is assigned at the task's row, and the passed-over seats + // are completed on that same row: set directly, because a completion + // applied as an event must land strictly above its assignment, and + // the task's row may be the log's first, at sequence zero. let mut episode = CompletionEpisodeState::opened( Conversation { desk_id: door.chat.clone(), desk_name: door.desk_name.clone(), thread_root: None, }, - Sequence(0), + door.opened_at, 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)?; + for participant in &mut episode.participants { + if !door.starters.contains(&participant.agent_id) { + for record in &mut participant.assignments { + record.completed_at = Some(door.opened_at); + } } } let state = driver.start(episode)?; @@ -301,13 +307,16 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { if !taken.insert(seat.clone()) { continue; } + // The ask row is the first thing a seat is shown in the + // conversation it roots: a first turn there starts just + // below it, which for a root at zero is nowhere. let since = child .state .seen() .delivered_through .get(&seat) .copied() - .unwrap_or(child.root); + .or_else(|| child.root.0.checked_sub(1).map(Sequence)); turns.push(Turn { channel: Channel::Thread { root: child.root, @@ -323,13 +332,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { if !taken.insert(seat.clone()) { continue; } - let since = self - .state - .seen() - .delivered_through - .get(&seat) - .copied() - .unwrap_or(Sequence(0)); + let since = self.state.seen().delivered_through.get(&seat).copied(); turns.push(Turn { seat, channel: Channel::Desk, @@ -356,21 +359,24 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { } /// Open a turn: record that the seat is shown everything through - /// `latest`, and that it ran for what it holds, and build its brief. + /// `latest` -- the log's newest row, or `None` for a log with none -- + /// 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, + latest: Option, 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); + if let Some(latest) = latest { + child.state.delivered(&turn.seat, latest); + } child.state.turn_started(&turn.seat); child.turns += 1; child.turned = true; @@ -393,7 +399,9 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { ) } Channel::Desk => { - self.state.delivered(&turn.seat, latest); + if let Some(latest) = latest { + self.state.delivered(&turn.seat, latest); + } self.state.turn_started(&turn.seat); let views = self.views(&turn.seat, &mut transcript); EpisodeBrief::for_turn( @@ -408,6 +416,26 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { } } + /// The conversations `seat` would be shown on its next desk turn, by + /// root: those concluded since it last spoke, and any still in progress. + /// A host that reads its log asynchronously fetches these transcripts + /// before [`open_turn`](Self::open_turn), which reads them by root. + #[must_use] + pub fn shown_conversations(&self, seat: &str) -> Vec { + let cursor = self.shown.get(seat).copied().unwrap_or(0); + self.concluded[cursor..] + .iter() + .filter(|done| done.involves(seat)) + .map(|done| done.root) + .chain( + self.children + .values() + .filter(|child| child.involves(seat)) + .map(|child| child.root), + ) + .collect() + } + /// 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( diff --git a/crates/tinyhivemind-driver/src/conduct/steps.rs b/crates/tinyhivemind-driver/src/conduct/steps.rs index 96a3f0ff..63ee9c67 100644 --- a/crates/tinyhivemind-driver/src/conduct/steps.rs +++ b/crates/tinyhivemind-driver/src/conduct/steps.rs @@ -20,9 +20,13 @@ pub struct Turn { 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, + /// The newest row the seat has been shown in this channel, or `None` + /// for a seat shown nothing there yet: the host gives the turn every + /// row above it, which for `None` is every row. A sequence is never + /// borrowed to mean "nothing": a host may number its first row zero. + /// On the wire the field is present, `null` for `None`. + #[serde(deserialize_with = "required_null")] + pub since: Option, } impl Turn { @@ -218,3 +222,14 @@ pub enum Step { /// Show this, or don't. Event(Event), } + +/// Deserialize a nullable field that must be present: `serde` fills a +/// missing `Option` with `None` by default, and a wire form that dropped +/// the field would then pass as one that sent `null`. +fn required_null<'de, D, T>(deserializer: D) -> std::result::Result, D::Error> +where + D: serde::Deserializer<'de>, + T: serde::Deserialize<'de>, +{ + Option::::deserialize(deserializer) +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs index 593d5d3a..d82ad7d0 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs @@ -52,7 +52,11 @@ fn an_ask_opens_a_conversation_that_runs_first_and_concludes_to_the_asker() { ("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_eq!( + answered.turns[0].since, + Some(Sequence(root.0 - 1)), + "the ask row itself is new to the seat asked" + ); assert!(matches!( answered.turns[0].channel, Channel::Thread { root: at, ref other, opened_it: false } if at == root && other == "one" @@ -72,6 +76,8 @@ fn an_ask_opens_a_conversation_that_runs_first_and_concludes_to_the_asker() { // The asker is released: it runs on the desk, is shown the whole // conversation once, and completes. + assert_eq!(conductor.shown_conversations("one"), vec![root]); + assert!(conductor.shown_conversations("three").is_empty()); let turns = conductor.turns().expect("turns"); let brief = conductor.open_turn(&turns[0], journal.latest(), Vec::new(), |root| { journal.thread(root) @@ -287,7 +293,7 @@ fn a_refused_reply_in_a_conversation_is_not_its_answer() { .iter() .find(|turn| turn.seat == "two") .expect("the askee is due"); - conductor.open_turn(thread_turn, Sequence(1), Vec::new(), |_| Vec::new()); + conductor.open_turn(thread_turn, Some(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") { diff --git a/crates/tinyhivemind-driver/src/conduct/test/door.rs b/crates/tinyhivemind-driver/src/conduct/test/door.rs index 5abacf0c..dc0d6172 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/door.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/door.rs @@ -6,6 +6,7 @@ use super::support::{ClarifyRouter, Journal, complete, door, hive, policy, run, use crate::conduct::{ConductPolicy, Conductor, starters}; use crate::driver::BroadcastRouting; use crate::{CompletionDriver, Error}; +use tinyhivemind::Sequence; use tinyhivemind_embed::{Router, RoutingFallback, RoutingPlan, RoutingRequest}; #[test] @@ -47,6 +48,37 @@ fn the_door_starts_the_routed_seats_and_completes_the_rest() { assert!(conductor.state().quiescent()); } +#[test] +fn a_task_at_sequence_zero_opens_the_episode_and_is_new_to_the_starter() { + 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: &[], + }; + // A host numbering from zero: the task is row zero, so the passed-over + // seat is completed on row zero, which no completion event could land on. + let journal = Journal::numbered_from(0); + let entrance = door(&["one", "two"], &["one"], &journal); + assert_eq!(entrance.opened_at, Sequence(0)); + let mut conductor = + Conductor::open(&driver, routing, ConductPolicy::default(), entrance).expect("opens"); + let turns = conductor.turns().expect("turns"); + assert_eq!(seats(&turns), vec![("one", None)]); + assert_eq!( + turns[0].since, None, + "shown nothing yet: row zero is above the watermark, not on it" + ); + let first = wave(&mut conductor, &journal, &[("one", vec![complete("done")])]).expect("wave"); + assert_eq!(first.turns.len(), 1); + assert!(conductor.finished()); + assert_eq!(journal.bodies(), vec!["the task", "COMPLETE: done"]); +} + #[test] fn the_door_refuses_a_starter_outside_the_desk_and_no_starter_at_all() { let hive = hive(&["one", "two"]); diff --git a/crates/tinyhivemind-driver/src/conduct/test/support.rs b/crates/tinyhivemind-driver/src/conduct/test/support.rs index 46a56153..6d8b8f3a 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/support.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/support.rs @@ -126,12 +126,29 @@ impl Router for ClarifyRouter { pub(super) type Row = (Sequence, String, String, Option, Option); /// The host: rows, and nothing else. -#[derive(Debug, Default)] +#[derive(Debug)] pub(super) struct Journal { + /// The sequence the first row is given. + first: u64, rows: Mutex>, } +impl Default for Journal { + fn default() -> Self { + Self::numbered_from(1) + } +} + impl Journal { + /// A journal whose first row is given `first`: some hosts number from + /// zero. + pub(super) fn numbered_from(first: u64) -> Self { + Self { + first, + rows: Mutex::new(Vec::new()), + } + } + pub(super) fn append( &self, author: &str, @@ -140,17 +157,13 @@ impl Journal { only_for: Option, ) -> Sequence { let mut rows = self.rows.lock().unwrap(); - let sequence = Sequence(rows.last().map_or(0, |row| row.0.0) + 1); + let sequence = Sequence(rows.last().map_or(self.first, |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 latest(&self) -> Option { + self.rows.lock().unwrap().last().map(|row| row.0) } pub(super) fn thread(&self, root: Sequence) -> Vec { diff --git a/crates/tinyhivemind-driver/src/conduct/test/wire.rs b/crates/tinyhivemind-driver/src/conduct/test/wire.rs index c40601ea..c0ed6a66 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/wire.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/wire.rs @@ -41,7 +41,7 @@ fn a_turn_names_its_seat_channel_and_watermark() { other: "one".into(), opened_it: false, }, - since: Sequence(4), + since: Some(Sequence(4)), }; let wire = serde_json::to_value(&thread).expect("serializes"); assert_eq!( @@ -58,11 +58,14 @@ fn a_turn_names_its_seat_channel_and_watermark() { let desk = Turn { seat: "one".into(), channel: Channel::Desk, - since: Sequence(0), + since: None, }; + let wire = serde_json::to_value(&desk).expect("serializes"); + assert_eq!(wire["channel"], json!({"kind": "desk"})); assert_eq!( - serde_json::to_value(&desk).expect("serializes")["channel"], - json!({"kind": "desk"}) + wire["since"], + json!(null), + "nothing shown yet is null on the wire, never a sequence" ); round_trips(&desk); } diff --git a/crates/tinyhivemind-openhuman/Cargo.toml b/crates/tinyhivemind-openhuman/Cargo.toml index 206b3804..fce26b40 100644 --- a/crates/tinyhivemind-openhuman/Cargo.toml +++ b/crates/tinyhivemind-openhuman/Cargo.toml @@ -49,7 +49,7 @@ serde_json.workspace = true thiserror.workspace = true # A turn has a wall; `tokio::time::timeout` is it. Both harnesses run on # tokio already. -tokio = { workspace = true, features = ["sync", "time"] } +tokio = { workspace = true, features = ["rt", "sync", "time"] } tinyhivemind.workspace = true # The record every call lands in, and the served definitions. tinyhivemind-tools.workspace = true @@ -61,6 +61,8 @@ wiremock = { workspace = true, optional = true } [dev-dependencies] tempfile.workspace = true +# The routing surfaces, for a test that builds its own hive. +tinyhivemind-embed.workspace = true tokio = { workspace = true, features = ["rt-multi-thread", "macros", "time"] } wiremock.workspace = true diff --git a/crates/tinyhivemind-openhuman/README.md b/crates/tinyhivemind-openhuman/README.md index 7cd6988c..d70786b5 100644 --- a/crates/tinyhivemind-openhuman/README.md +++ b/crates/tinyhivemind-openhuman/README.md @@ -10,7 +10,10 @@ This crate is the host's side of that seam for OpenHuman, three ways: | `EmbedRunner` | an `openhuman-embed` `AgentSpec` agent on a runtime the host booted | the three MCP dispatchers, dialling `tinyhivemind-mcp`'s server | OpenHuman's own session, stable for the episode | | `RawRunner` | an `OpenHumanSessionHost` built one level down, per turn | the same tools in-process, each calling `EpisodeTools::call` | a per-seat log this crate seeds the next session with | -All three implement `SeatRunner`, the seam: open a turn, run it, close it and +`run_episode` runs one episode from its door to quiescence over any of them +and a `Journal` the host implements -- its log, and how it appends the +conductor's rows -- so a host builds a driver, a door and a runner and calls +one function. All three implement `SeatRunner`, the seam: open a turn, run it, close it and take what was called. Open and close are the same for every runner, because every call lands in the same `EpisodeTools`, so the driver drains identical events and a seat is refused and acknowledged in the same words whichever diff --git a/crates/tinyhivemind-openhuman/src/README.md b/crates/tinyhivemind-openhuman/src/README.md index ee8850ad..e991f59f 100644 --- a/crates/tinyhivemind-openhuman/src/README.md +++ b/crates/tinyhivemind-openhuman/src/README.md @@ -2,10 +2,12 @@ | Path | Purpose | |---|---| -| `lib.rs` | Crate overview and the public surface: `SeatRunner`, `RunnerKind`, `HostedRunner`, `EpisodeHost`, `EpisodeBelt`, `EmbedRunner`, `EmbedSeat`, `RawRunner`, `RawSeat`, `LibraryHost`, `Route`, `register_seats`, `offline`. | -| `error/` | What seating or running a seat can fail with. | +| `lib.rs` | Crate overview and the public surface: `run_episode`, `Journal`, `Report`, `SeatRunner`, `RunnerKind`, `HostedRunner`, `EpisodeHost`, `EpisodeBelt`, `EmbedRunner`, `EmbedSeat`, `RawRunner`, `RawSeat`, `LibraryHost`, `Route`, `register_seats`, `offline`. | +| `error/` | What seating or running a seat, or an episode, can fail with. | +| `episode/` | `run_episode` over a `Journal`: one episode from its door to quiescence. | | `runner/` | The seam: open, run, close; `Lane`, `TurnJob`; which runner the environment names. | +| `journal/` | `MemoryLog`, an in-memory journal that is a real `SessionLog`; always compiled. | | `hosted/` | Seats as the host's own agents, built through `EpisodeHost`, seeded from the host's log. | | `embed/` | Seats as `openhuman-embed` agents, tools over MCP. | | `raw/` | Seats as raw sessions, tools in-process: the belt, the gate, the memory that keeps nothing. | -| `offline/` | The scripted model, the backend stub, the offline config and an in-memory journal that is a real `SessionLog`, behind the `offline` feature and in tests. | +| `offline/` | The scripted model, the backend stub and the offline config, behind the `offline` feature and in tests. | diff --git a/crates/tinyhivemind-openhuman/src/embed/mod.rs b/crates/tinyhivemind-openhuman/src/embed/mod.rs index 3c6939a7..d2d7bcbc 100644 --- a/crates/tinyhivemind-openhuman/src/embed/mod.rs +++ b/crates/tinyhivemind-openhuman/src/embed/mod.rs @@ -120,7 +120,7 @@ impl SeatRunner for EmbedRunner { /// One session per seat for the whole episode, so `OpenHuman` appends to the /// context the agent already holds rather than rebuilding one. - fn turn(&self, seat: String, lane: Lane, _since: Sequence, prompt: String) -> TurnJob { + fn turn(&self, seat: String, lane: Lane, _since: Option, prompt: String) -> TurnJob { let Some(agent) = self.agents.get(&seat).cloned() else { return unseated(seat, lane); }; diff --git a/crates/tinyhivemind-openhuman/src/episode/README.md b/crates/tinyhivemind-openhuman/src/episode/README.md new file mode 100644 index 00000000..efcf7e44 --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/README.md @@ -0,0 +1,31 @@ +# `episode` + +`run_episode`: one episode from its door to quiescence, over a journal the +host owns. It moves rows between the host's log, the runner and the +conductor, one wave at a time -- begin the wave, propose the turns, brief +and run each, record what each called, then take the conductor's steps +until the wave settles -- and holds no rule of its own; every rule is the +conductor's. + +`Journal` is what a host implements: its `SessionLog`, `commit` and `note` +to append the conductor's rows and return the sequence a commit was given, +and three optional hooks -- `event` to show what the episode did, `compose` +to put its own context in front of the brief, `turn_done` to see a turn's +reply, refusals and recorded calls. `Report` is what an episode came to. + +Rows for a turn are read from the log through `project_session`, as the +seat, so a row it was not addressed on is withheld the same way it is when +the turn is seeded. One watermark is read per wave, the log's newest row, +and every read for the wave is bounded by it: the host's log may grow while +the turns are prepared, and a seat is recorded as shown through the +watermark, so a row above it waits for the next turn rather than being +shown twice. The rows above the seat's own watermark are its brief; the +conversations it is shown are fetched by root from +`Conductor::shown_conversations`. A seat shown nothing yet has no +watermark, and a sequence is never borrowed to mean that: a host may number +its first row zero. + +| file | holds | +| --- | --- | +| `mod.rs` | `Journal`, `Report`, `run_episode`, reading and rendering rows | +| `test/` | the loop over a scripted runner and no model: `flow.rs` (an episode with a conversation, a stalled one, what the journal saw of each), `watermark.rs` (a log numbered from zero, a log that grows under the loop), `journals.rs` (a journal keeping every default), `support.rs` (the runner and the journals) | diff --git a/crates/tinyhivemind-openhuman/src/episode/mod.rs b/crates/tinyhivemind-openhuman/src/episode/mod.rs new file mode 100644 index 00000000..8bfa50f9 --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/mod.rs @@ -0,0 +1,316 @@ +//! One episode, from its door to quiescence, over a journal the host owns. +//! +//! The [`Conductor`] holds every rule of a completion episode and the +//! [`SeatRunner`] runs a turn; between them sits the loop that moves rows +//! from the host's journal to a seat and back, one wave at a time: +//! +//! 1. `begin_wave`: the nudges due, appended. +//! 2. `turns`: who runs, where, above which row. +//! 3. For each turn: the rows the seat has not seen, read from the log as +//! the seat; the record opened for the turn; the brief; the prompt the +//! host composes; the turn started on the runner. +//! 4. Every turn awaited together, closed, and what it called recorded. +//! 5. `step` until settled: a commit appended and its sequence reported, a +//! note appended, an event shown. +//! +//! [`run_episode`] is that loop. It reads through the host's [`SessionLog`] +//! and writes through [`Journal`], so a host implements the journal and +//! calls one function. + +#[cfg(test)] +mod test; + +use tinyhivemind::aside::Viewer; +use tinyhivemind::{ + Conversation, SESSION_WINDOW, Sequence, SessionAuthor, SessionLog, SessionMessage, + SessionQuery, project_session, +}; +use tinyhivemind_driver::{ + BoundAgent, BroadcastRouting, Commit, CompletionDriver, ConductPolicy, Conductor, Door, + EpisodeBrief, Event, Note, Step, +}; +use tinyhivemind_tools::{Dispatch, Refusal}; + +use crate::Result; +use crate::runner::{Lane, SeatRunner, TurnJob, TurnResult}; + +/// The journal an episode runs over: what the host reads for a seat and +/// appends on the conductor's behalf, and what it wants to see of a turn. +/// +/// The host renders its own rows. A [`Commit`] carries the utterance and the +/// host appends it as a row of its own shape, returning the sequence the +/// journal gave it; a [`Note`] carries the desk's words, attributed to the +/// desk. Everything else has a default. +pub trait Journal: Send + Sync { + /// The host's log, read as a seat to seed and to brief a turn. + fn log(&self) -> &dyn SessionLog; + + /// Append what a seat said, or what the episode says on its behalf, and + /// return the sequence it was given. + /// + /// # Errors + /// + /// The journal refusing the row. + fn commit(&self, commit: &Commit) -> Result; + + /// Append what the desk says to a seat. + /// + /// # Errors + /// + /// The journal refusing the row. + fn note(&self, note: &Note) -> Result<()>; + + /// Something the episode did, to show or not. The default shows nothing. + fn event(&self, event: &Event) { + let _ = event; + } + + /// The message a turn is sent. The default is the brief as the episode + /// words it; a host prepends what it owns. + fn compose(&self, seat: &str, brief: &EpisodeBrief) -> String { + let _ = seat; + brief.render() + } + + /// A turn came back: its reply or failure, what it was refused, and how + /// many calls it made that the record accepted. The default does nothing. + fn turn_done( + &self, + seat: &str, + lane: Lane, + outcome: &TurnResult, + refused: &[Refusal], + recorded: usize, + ) { + let _ = (seat, lane, outcome, refused, recorded); + } +} + +/// What one episode came to. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct Report { + /// Seat turns run. + pub turns: u64, + /// Waves proposed. + pub waves: u64, + /// Seats completed with their work for a spent broadcast budget. + pub discharged: u64, + /// Conversations concluded. + pub conversations: usize, + /// Seats settled at the end. + pub settled: usize, +} + +/// Run one episode from `door` to quiescence. +/// +/// # Errors +/// +/// The journal refusing a row, the runner failing to open a turn, or the +/// conductor stopping the episode: a stalled desk, a wall, or a fold error +/// it could not explain to the seat. +pub async fn run_episode( + journal: &J, + runner: &R, + driver: &CompletionDriver<'_, A>, + routing: BroadcastRouting<'_>, + policy: ConductPolicy, + door: Door, +) -> Result +where + A: BoundAgent, + J: Journal, + R: SeatRunner, +{ + let desk = Conversation { + desk_id: door.chat.clone(), + desk_name: door.desk_name.clone(), + thread_root: None, + }; + let mut conductor = Conductor::open(driver, routing, policy, door)?; + loop { + if conductor.finished() { + break; + } + for step in conductor.begin_wave() { + settle(journal, &mut conductor, step).await?; + } + let turns = conductor.turns()?; + // One watermark for the wave, and every read bounded by it: the + // host's log may grow while the turns are prepared, and a row above + // the watermark shown now would be shown again next turn, since a + // seat is recorded as shown through the watermark and no further. + let latest = latest(journal.log()).await?; + let mut jobs: Vec = Vec::with_capacity(turns.len()); + for turn in &turns { + let channel = Conversation { + thread_root: turn.thread(), + ..desk.clone() + }; + let rows = rows_above(journal.log(), &channel, &turn.seat, turn.since, latest).await?; + let window = match turn.thread() { + None => rows.clone(), + Some(_) => rows_above(journal.log(), &channel, &turn.seat, None, latest).await?, + }; + runner.open( + &turn.seat, + window, + Dispatch { + chat: desk.desk_id.clone(), + parent: turn.thread().map(|root| root.0.to_string()), + }, + ); + // Only a desk turn is shown its conversations, so only a desk + // turn reads them. + let mut transcripts = std::collections::BTreeMap::new(); + let shown = match turn.thread() { + None => conductor.shown_conversations(&turn.seat), + Some(_) => Vec::new(), + }; + for root in shown { + let thread = Conversation { + thread_root: Some(root), + ..desk.clone() + }; + let whole = rows_above(journal.log(), &thread, &turn.seat, None, latest).await?; + transcripts.insert(root, whole); + } + let brief = conductor.open_turn(turn, latest, rows, |root| { + transcripts.get(&root).cloned().unwrap_or_default() + }); + let prompt = journal.compose(&turn.seat, &brief); + let lane = turn.thread().map_or(Lane::Desk, Lane::Thread); + jobs.push(runner.turn(turn.seat.clone(), lane, turn.since, prompt)); + } + let named: Vec<(String, Lane)> = turns + .iter() + .map(|turn| { + ( + turn.seat.clone(), + turn.thread().map_or(Lane::Desk, Lane::Thread), + ) + }) + .collect(); + for (seat, lane, outcome) in join_turns(jobs, named).await { + // Close the turn first: the record refuses a call on a closed + // turn, so nothing can land after this point is read. + let events = runner.close(&seat); + let refused = runner.tools().drain_refusals(&seat); + journal.turn_done(&seat, lane, &outcome, &refused, events.len()); + if let Some(turn) = turns.iter().find(|turn| turn.seat == seat) { + conductor.record(turn, events.into_iter().map(|event| event.call)); + } + } + while let Some(step) = conductor.step()? { + settle(journal, &mut conductor, step).await?; + } + } + Ok(Report { + turns: conductor.turns_run(), + waves: conductor.waves(), + discharged: conductor.discharged(), + conversations: conductor.conversations(), + settled: conductor.state().episode().settled(), + }) +} + +/// One step taken: a commit appended and its sequence reported, a note +/// appended, an event shown. +async fn settle( + journal: &J, + conductor: &mut Conductor<'_, A>, + step: Step, +) -> Result<()> { + match step { + Step::Commit(commit) => { + let sequence = journal.commit(&commit)?; + conductor.committed(sequence).await?; + } + Step::Note(note) => journal.note(¬e)?, + Step::Event(event) => journal.event(&event), + } + Ok(()) +} + +/// Every turn of a wave, run together, in the order they finish. A turn +/// whose task panicked is a failed turn, not a failed wave: `named` says +/// which seat and lane each job was, in the order the jobs were made. +async fn join_turns( + jobs: Vec, + named: Vec<(String, Lane)>, +) -> Vec<(String, Lane, TurnResult)> { + let mut tasks = tokio::task::JoinSet::new(); + let mut who = std::collections::HashMap::new(); + for (job, name) in jobs.into_iter().zip(named) { + who.insert(tasks.spawn(job).id(), name); + } + let mut done = Vec::new(); + while let Some(joined) = tasks.join_next_with_id().await { + match joined { + Ok((_, outcome)) => done.push(outcome), + Err(error) => { + let (seat, lane) = who + .remove(&error.id()) + .unwrap_or_else(|| (String::new(), Lane::Desk)); + done.push(( + seat, + lane, + Some(Err(format!("the turn's task failed: {error}"))), + )); + } + } + } + done +} + +/// The newest sequence in the log, or `None` for a log with no rows. +async fn latest(log: &dyn SessionLog) -> Result> { + let page = log + .read_before(None, 1) + .await + .map_err(|source| tinyhivemind::Error::Read { source })?; + Ok(page.messages.first().map(|row| row.sequence)) +} + +/// The rows of `conversation` above `since` and through `latest` that +/// `seat` may read, rendered, newest [`SESSION_WINDOW`] of them: every row +/// for a `since` of `None`, and none for a `latest` of `None`, the wave's +/// watermark on a log that had no rows. +async fn rows_above( + log: &dyn SessionLog, + conversation: &Conversation, + seat: &str, + since: Option, + latest: Option, +) -> Result> { + let Some(latest) = latest else { + return Ok(Vec::new()); + }; + let rows = project_session( + log, + &SessionQuery { + conversation: conversation.clone(), + viewer: Viewer::Agent { id: seat.into() }, + // Exclusive, so one above the watermark; nothing is above the + // last sequence, so that reads unbounded rather than one short. + before: latest.0.checked_add(1).map(Sequence), + window: SESSION_WINDOW, + }, + ) + .await?; + Ok(rows + .iter() + .filter(|row| since.is_none_or(|since| row.sequence > since)) + .filter_map(render) + .collect()) +} + +/// `@author: content`, or nothing for a row the seat may not read. +fn render(row: &SessionMessage) -> Option { + let content = row.readable()?; + let author = match &row.author { + SessionAuthor::Operator => "operator", + SessionAuthor::Agent { id, .. } => id, + SessionAuthor::Person { label, .. } | SessionAuthor::System { label, .. } => label, + }; + Some(format!("@{author}: {content}")) +} diff --git a/crates/tinyhivemind-openhuman/src/episode/test/README.md b/crates/tinyhivemind-openhuman/src/episode/test/README.md new file mode 100644 index 00000000..0407e5aa --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/README.md @@ -0,0 +1,11 @@ +# `episode/test` + +The loop over a scripted runner and no model. + +| file | holds | +| --- | --- | +| `mod.rs` | the module doc and the submodules | +| `support.rs` | `ScriptRunner`, whose turns are scripted tool calls; `TestJournal`, which records what it was shown; `BareJournal` and `GrowingLog`; the desk, the door and the calls | +| `flow.rs` | an episode with a conversation and one that stalls, and what the journal and the runner saw of each | +| `watermark.rs` | a task on a log numbered from zero, and a row the host appends above a wave's watermark | +| `journals.rs` | a journal that keeps every default is briefed as the episode words it | diff --git a/crates/tinyhivemind-openhuman/src/episode/test/flow.rs b/crates/tinyhivemind-openhuman/src/episode/test/flow.rs new file mode 100644 index 00000000..2d883397 --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/flow.rs @@ -0,0 +1,203 @@ +//! An episode from its door to quiescence, and one that stalls: what the +//! journal and the runner saw of each. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use serde_json::json; +use tinyhivemind::Sequence; +use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Event}; + +use super::super::{Report, run_episode}; +use super::support::{ScriptRunner, TestJournal, ask, complete, door, hive, policy, post, run}; +use crate::Error; +use crate::runner::Lane; + +#[test] +fn an_episode_runs_from_its_door_to_quiescence_over_the_journal() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = TestJournal::new(); + // One asks two (row 2 roots the conversation) and, woken by its own ask + // row, says nothing; two tries a post, which is not served, and answers + // in the thread with a completion; one, released by the conclusion, + // completes on the desk. + let runner = ScriptRunner::new( + &["one", "two"], + &[ + ( + "one", + vec![ + vec![ask("two", "which port?", None)], + vec![], + vec![complete("fixed", None)], + ], + ), + ( + "two", + vec![vec![post("checking", 2), complete("port 8080", Some(2))]], + ), + ], + ); + let report = run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + door(&journal, &["one", "two"], &["one"]), + )) + .expect("the episode settles"); + assert_eq!( + report, + Report { + turns: 4, + waves: 3, + discharged: 0, + conversations: 1, + settled: 2, + }, + "{:?}", + journal.log.all() + ); + let bodies: Vec = journal + .log + .all() + .iter() + .map(|row| row.body.clone()) + .collect(); + assert_eq!(bodies[0], "state the root cause"); + assert!(bodies.contains(&"which port?".to_owned())); + assert!(bodies.contains(&"port 8080".to_owned())); + assert!( + bodies + .iter() + .any(|body| body.contains("concluded our conversation")) + ); + assert!(bodies.contains(&"fixed".to_owned())); + let events = journal.events(); + assert!( + events + .iter() + .any(|event| matches!(event, Event::Asked { root, .. } if *root == Sequence(2))) + ); + assert!( + events + .iter() + .any(|event| matches!(event, Event::Concluded { forced: false, .. })) + ); + + the_journal_saw_each_turn(&journal, &runner); +} + +/// The thread turn was briefed with the thread, the asker's waking desk +/// turn was shown the concluded conversation, and every turn came back to +/// the journal with what it recorded. +fn the_journal_saw_each_turn(journal: &TestJournal, runner: &ScriptRunner) { + let prompts = runner.prompts(); + let (_, lane, thread_prompt) = prompts + .iter() + .find(|(seat, _, _)| seat == "two") + .expect("two ran"); + assert_eq!(*lane, Lane::Thread(Sequence(2))); + assert!(thread_prompt.contains("which port?"), "{thread_prompt}"); + let shown = journal.shown.lock().unwrap(); + assert!( + shown + .iter() + .any(|(seat, conversations)| seat == "one" && *conversations == 1) + ); + // Every turn came back to the journal with what it recorded. + let turns = journal.turns.lock().unwrap(); + assert_eq!(turns.len(), 4, "three that called, and one's silent turn"); + assert!( + turns + .iter() + .all(|(_, _, outcome, _, _)| matches!(outcome, Some(Ok(_)))) + ); + // `post` is in the vocabulary and not served: two's post in the thread + // was refused inside its turn, and the journal was told so. + let refusals: Vec<(&str, usize)> = turns + .iter() + .map(|(seat, _, _, refused, _)| (seat.as_str(), *refused)) + .filter(|(_, refused)| *refused > 0) + .collect(); + assert_eq!(refusals, vec![("two", 1)]); + assert_eq!( + turns + .iter() + .filter(|(_, _, _, _, recorded)| *recorded >= 1) + .count(), + 3 + ); +} + +#[test] +fn a_seat_that_says_nothing_is_nudged_and_then_the_episode_stalls() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = TestJournal::new(); + let runner = ScriptRunner::new( + &["one", "two"], + &[("one", vec![vec![], vec![("fail", json!({}))]])], + ); + let stalled = run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + door(&journal, &["one", "two"], &["one"]), + )); + assert!( + matches!(&stalled, Err(Error::Conduct(tinyhivemind_driver::Error::Stalled { seats })) if seats == &["one".to_owned()]), + "{stalled:?}" + ); + // The nudge reached the journal as a row to one alone, and the failed + // turn reached it as a turn that failed. + let rows = journal.log.all(); + assert!( + rows.iter() + .any(|row| row.author == "desk" && row.only_for.as_deref() == Some("one")) + ); + let turns = journal.turns.lock().unwrap(); + assert!( + turns + .iter() + .any(|(_, _, outcome, _, recorded)| matches!(outcome, Some(Err(_))) && *recorded == 0) + ); + assert!( + journal + .events() + .iter() + .any(|event| matches!(event, Event::Nudged { thread: None, .. })) + ); + // The nudge is one's alone: it woke nobody else, and what two would be + // shown of the desk does not hold it. + let prompts = runner.prompts(); + assert!( + prompts.iter().all(|(seat, _, _)| seat == "one"), + "{prompts:?}" + ); + let shown_two = journal.log.desk_since("two", None); + assert!(!shown_two.is_empty()); + assert!( + shown_two.iter().all(|row| !row.contains("open work")), + "{shown_two:?}" + ); +} diff --git a/crates/tinyhivemind-openhuman/src/episode/test/journals.rs b/crates/tinyhivemind-openhuman/src/episode/test/journals.rs new file mode 100644 index 00000000..dea9895e --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/journals.rs @@ -0,0 +1,63 @@ +//! A journal that keeps every default is briefed as the episode words it. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use serde_json::json; +use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Door}; + +use super::super::run_episode; +use super::support::{BareJournal, ScriptRunner, complete, hive, policy, run}; +use crate::journal::MemoryLog; + +#[test] +fn a_journal_that_keeps_the_defaults_is_briefed_as_the_episode_words_it() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = BareJournal(MemoryLog::new("engineering")); + let opened_at = journal.0.append("operator", "the task", None, None); + // The first turn's task panics; the second completes. A panicked task + // is a failed turn, not a failed wave, and the seat runs again. + let runner = ScriptRunner::new( + &["one", "two"], + &[( + "one", + vec![vec![("panic", json!({}))], vec![complete("done", None)]], + )], + ); + let report = run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + Door { + chat: "engineering".into(), + desk_name: "Engineering".into(), + members: vec!["one".into(), "two".into()], + starters: vec!["one".into()], + opened_at, + }, + )) + .expect("the episode settles"); + assert_eq!(report.settled, 2); + assert_eq!(report.conversations, 0); + // The default composition is the brief alone: the operator's row, as the + // episode renders it. + let prompts = runner.prompts(); + assert!(prompts[0].2.starts_with("## "), "{}", prompts[0].2); + assert!( + prompts[0].2.contains("@operator: the task"), + "{}", + prompts[0].2 + ); + assert_eq!(prompts.len(), 2, "the panicked turn was run again"); + assert!(journal.0.all().iter().any(|row| row.body == "done")); +} diff --git a/crates/tinyhivemind-openhuman/src/episode/test/mod.rs b/crates/tinyhivemind-openhuman/src/episode/test/mod.rs new file mode 100644 index 00000000..30b2af4b --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/mod.rs @@ -0,0 +1,6 @@ +//! The episode loop over a journal, with a scripted runner and no model. + +mod flow; +mod journals; +mod support; +mod watermark; diff --git a/crates/tinyhivemind-openhuman/src/episode/test/support.rs b/crates/tinyhivemind-openhuman/src/episode/test/support.rs new file mode 100644 index 00000000..6d344755 --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/support.rs @@ -0,0 +1,367 @@ +//! Fixtures: a scripted runner, journals over the in-memory log, and the +//! desk every test opens. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use std::collections::{BTreeMap, VecDeque}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, PoisonError}; + +use serde_json::{Value, json}; +use tinyhivemind::desk::{Desk, ResponderMode}; +use tinyhivemind::responder::Probability; +use tinyhivemind::{Sequence, SessionFuture, SessionLog}; +use tinyhivemind_driver::{ + AgentBinding, BoundAgent, BoundHive, Commit, Door, EpisodeBrief, Event, HiveGraph, Note, +}; +use tinyhivemind_embed::{RouteCandidate, RoutingPolicy}; +use tinyhivemind_tools::{EpisodeTools, Refusal}; + +use super::super::Journal; +use crate::MemoryLog; +use crate::Result; +use crate::runner::{Lane, SeatRunner, TurnJob, TurnResult}; + +/// A seat with nothing behind it. +#[derive(Clone, Debug)] +pub(super) struct Seat(String); + +impl BoundAgent for Seat { + fn runtime_id(&self) -> &str { + &self.0 + } +} + +/// One scripted call: a tool by its served name, and its arguments. +pub(super) type Call = (&'static str, Value); + +/// A runner whose seats say what they were told to, turn by turn, straight +/// into the record: what a model would do, without one. +pub(super) struct ScriptRunner { + tools: Arc, + seats: Vec, + script: Mutex>>>, + /// Every prompt a seat was sent, in order. + prompts: Mutex>, + /// The watermark each turn was opened with, in order. + since: Mutex>>, +} + +impl ScriptRunner { + pub(super) fn new(seats: &[&str], script: &[(&str, Vec>)]) -> Self { + Self { + tools: Arc::new(EpisodeTools::new(seats.iter().map(|id| (*id).to_owned()))), + seats: seats.iter().map(|id| (*id).to_owned()).collect(), + script: Mutex::new( + script + .iter() + .map(|(seat, turns)| ((*seat).to_owned(), turns.clone().into())) + .collect(), + ), + prompts: Mutex::new(Vec::new()), + since: Mutex::new(Vec::new()), + } + } + + pub(super) fn prompts(&self) -> Vec<(String, Lane, String)> { + self.prompts + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } + + pub(super) fn since(&self) -> Vec> { + self.since + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } +} + +impl SeatRunner for ScriptRunner { + fn tools(&self) -> &Arc { + &self.tools + } + + type Bound = Seat; + + fn bindings(&self) -> Vec> { + self.seats + .iter() + .map(|id| AgentBinding::new(id.clone(), Seat(id.clone()))) + .collect() + } + + fn turn(&self, seat: String, lane: Lane, since: Option, prompt: String) -> TurnJob { + self.prompts + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push((seat.clone(), lane, prompt)); + self.since + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(since); + let calls = self + .script + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get_mut(&seat) + .and_then(VecDeque::pop_front) + .unwrap_or_default(); + let tools = Arc::clone(&self.tools); + Box::pin(async move { + if calls.iter().any(|(name, _)| *name == "fail") { + return (seat, lane, Some(Err("the model went away".into()))); + } + assert!( + !calls.iter().any(|(name, _)| *name == "panic"), + "the model's task panicked" + ); + for (name, arguments) in &calls { + let _ = tools.call(&seat, name, arguments); + } + (seat, lane, Some(Ok("said".into()))) + }) + } +} + +/// What the journal saw of one turn: seat, lane, outcome, refusals, +/// recorded calls. +pub(super) type Seen = (String, Lane, TurnResult, usize, usize); + +/// A journal over the memory log that keeps what it was shown. +pub(super) struct TestJournal { + pub(super) log: MemoryLog, + pub(super) events: Mutex>, + pub(super) turns: Mutex>, + /// Each desk brief's conversations, by seat, as composed. + pub(super) shown: Mutex>, +} + +impl TestJournal { + pub(super) fn new() -> Self { + Self::over(MemoryLog::new("engineering")) + } + + pub(super) fn over(log: MemoryLog) -> Self { + Self { + log, + events: Mutex::new(Vec::new()), + turns: Mutex::new(Vec::new()), + shown: Mutex::new(Vec::new()), + } + } + + pub(super) fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } +} + +impl Journal for TestJournal { + fn log(&self) -> &dyn SessionLog { + &self.log + } + + fn commit(&self, commit: &Commit) -> Result { + Ok(self.log.append( + &commit.author, + commit.utterance.message(), + commit.thread, + commit.only_for.as_deref(), + )) + } + + fn note(&self, note: &Note) -> Result<()> { + self.log + .append("desk", ¬e.body, note.thread, note.only_for.as_deref()); + Ok(()) + } + + fn event(&self, event: &Event) { + self.events + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(event.clone()); + } + + fn compose(&self, seat: &str, brief: &EpisodeBrief) -> String { + self.shown + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push((seat.to_owned(), brief.conversations.len())); + format!("[{seat}]\n{}", brief.render()) + } + + fn turn_done( + &self, + seat: &str, + lane: Lane, + outcome: &TurnResult, + refused: &[Refusal], + recorded: usize, + ) { + self.turns + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(( + seat.to_owned(), + lane, + outcome.clone(), + refused.len(), + recorded, + )); + } +} + +pub(super) fn probability(parts: u32) -> Probability { + Probability::new(parts).expect("bounded") +} + +pub(super) fn policy() -> 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: 1, + 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") +} + +pub(super) fn door(journal: &TestJournal, ids: &[&str], starters: &[&str]) -> Door { + let opened_at = journal + .log + .append("operator", "state the root cause", 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 complete(message: &str, parent: Option) -> Call { + ( + "complete_episode", + json!({"message": message, "chat": "engineering", "parent": parent.map(|p| p.to_string())}), + ) +} + +pub(super) fn ask(to: &str, message: &str, parent: Option) -> Call { + ( + "ask", + json!({"to": to, "message": message, "chat": "engineering", "parent": parent.map(|p| p.to_string())}), + ) +} + +pub(super) fn post(message: &str, parent: u64) -> Call { + ( + "post", + json!({"message": message, "chat": "engineering", "parent": parent.to_string()}), + ) +} + +pub(super) fn run(future: F) -> F::Output { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime") + .block_on(future) +} + +pub(super) struct BareJournal(pub(super) MemoryLog); + +impl Journal for BareJournal { + fn log(&self) -> &dyn SessionLog { + &self.0 + } + + fn commit(&self, commit: &Commit) -> Result { + Ok(self.0.append( + &commit.author, + commit.utterance.message(), + commit.thread, + commit.only_for.as_deref(), + )) + } + + fn note(&self, note: &Note) -> Result<()> { + self.0 + .append("desk", ¬e.body, note.thread, note.only_for.as_deref()); + Ok(()) + } +} + +/// A log that grows under the loop: a host row lands right after a wave's +/// watermark is read, once. +pub(super) struct GrowingLog { + pub(super) inner: MemoryLog, + pub(super) late: AtomicBool, +} + +impl SessionLog for GrowingLog { + fn read_before(&self, before: Option, limit: usize) -> SessionFuture<'_> { + let page = self.inner.read_before(before, limit); + Box::pin(async move { + let page = page.await?; + if before.is_none() && limit == 1 && self.late.swap(false, Ordering::SeqCst) { + self.inner.append("operator", "one more thing", None, None); + } + Ok(page) + }) + } +} + +impl Journal for GrowingLog { + fn log(&self) -> &dyn SessionLog { + self + } + + fn commit(&self, commit: &Commit) -> Result { + Ok(self.inner.append( + &commit.author, + commit.utterance.message(), + commit.thread, + commit.only_for.as_deref(), + )) + } + + fn note(&self, note: &Note) -> Result<()> { + self.inner + .append("desk", ¬e.body, note.thread, note.only_for.as_deref()); + Ok(()) + } +} diff --git a/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs b/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs new file mode 100644 index 00000000..bb19444e --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs @@ -0,0 +1,108 @@ +//! The wave watermark: a log numbered from zero, and a log that grows under +//! the loop. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use std::sync::atomic::AtomicBool; + +use tinyhivemind::Sequence; +use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Door}; + +use super::super::run_episode; +use super::support::{GrowingLog, ScriptRunner, TestJournal, complete, door, hive, policy, run}; +use crate::journal::MemoryLog; + +#[test] +fn a_task_on_a_log_numbered_from_zero_reaches_the_first_turn() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + // A host that numbers its first row zero, as some do: the task is row + // zero, and nothing sits below it. + let journal = TestJournal::over(MemoryLog::numbered_from("engineering", Sequence(0))); + let runner = ScriptRunner::new( + &["one", "two"], + &[("one", vec![vec![complete("done", None)]])], + ); + let entrance = door(&journal, &["one", "two"], &["one"]); + assert_eq!(entrance.opened_at, Sequence(0)); + let report = run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + entrance, + )) + .expect("the episode runs"); + assert_eq!(report.settled, 2); + assert_eq!(report.turns, 1); + // The seat's one turn was shown the task, and was shown nothing before. + let prompts = runner.prompts(); + assert_eq!(prompts.len(), 1); + assert!( + prompts[0].2.contains("state the root cause"), + "{}", + prompts[0].2 + ); + assert_eq!(runner.since(), vec![None]); + // The completion landed above the task, at row one. + let rows = journal.log.all(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[1].sequence, Sequence(1)); + assert!(rows[1].body.contains("done")); +} + +#[test] +fn a_row_the_host_appends_above_the_wave_watermark_is_shown_once_and_later() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = GrowingLog { + inner: MemoryLog::new("engineering"), + late: AtomicBool::new(true), + }; + let opened_at = journal + .inner + .append("operator", "state the root cause", None, None); + let runner = ScriptRunner::new( + &["one", "two"], + &[("one", vec![vec![], vec![complete("done", None)]])], + ); + run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + Door { + chat: "engineering".into(), + desk_name: "Engineering".into(), + members: vec!["one".into(), "two".into()], + starters: vec!["one".into()], + opened_at, + }, + )) + .expect("the episode runs"); + // The late row is above the first wave's watermark: not in that turn, + // in the next, and in no more than one. + let prompts = runner.prompts(); + assert_eq!(prompts.len(), 2, "{prompts:?}"); + assert!(!prompts[0].2.contains("one more thing"), "{}", prompts[0].2); + assert!(prompts[1].2.contains("one more thing"), "{}", prompts[1].2); + assert_eq!(runner.since(), vec![None, Some(opened_at)]); +} diff --git a/crates/tinyhivemind-openhuman/src/error/mod.rs b/crates/tinyhivemind-openhuman/src/error/mod.rs index 3164a3e6..b435459d 100644 --- a/crates/tinyhivemind-openhuman/src/error/mod.rs +++ b/crates/tinyhivemind-openhuman/src/error/mod.rs @@ -19,8 +19,10 @@ pub enum Error { /// The id. seat: String, }, - /// A seat's definition was written and the registry did not load it. - #[error("seat `{seat}` did not register")] + /// A seat's definition was written and the registry did not load it: + /// the process registry is read once, so a seat written after that + /// first read is never seen. + #[error("seat `{seat}` did not register: the process registry was read before it was written")] SeatNotRegistered { /// The seat. seat: String, @@ -41,9 +43,13 @@ pub enum Error { #[error(transparent)] Agent(#[from] openhuman_embed::AgentError), /// The host's log failed to read, or broke the port's contract, while a - /// turn was being seeded. + /// turn was being seeded or briefed. #[error(transparent)] Session(#[from] tinyhivemind::Error), + /// The conductor stopped the episode: a stalled desk, a wall, or a fold + /// error it could not explain to the seat. + #[error(transparent)] + Conduct(#[from] tinyhivemind_driver::Error), /// `OpenHuman` refused: booting as a library host, resolving the route, /// building or seeding a session, or running the turn. #[error(transparent)] diff --git a/crates/tinyhivemind-openhuman/src/hosted/README.md b/crates/tinyhivemind-openhuman/src/hosted/README.md index 49895d45..684b2b5e 100644 --- a/crates/tinyhivemind-openhuman/src/hosted/README.md +++ b/crates/tinyhivemind-openhuman/src/hosted/README.md @@ -21,8 +21,8 @@ past its gate. The record is called by the served name either way. The definition a host registers for a seat must name the prefixed tools, since the hosted turn's allowlist comes from there (`register_seats`). -The host owns the log: `log()` borrows a `SessionLog` the host holds, over -its own journal, and the runner never keeps rows of its own. +A host is a `Journal` first: `log()` borrows a `SessionLog` the host holds +over its own journal, and the runner never keeps rows of its own. `after_turn` runs once a turn has run, with the usage the session reported. It is where a host parks what the turn left waiting on approval, meters the diff --git a/crates/tinyhivemind-openhuman/src/hosted/mod.rs b/crates/tinyhivemind-openhuman/src/hosted/mod.rs index 52ea39ee..bddd7cc7 100644 --- a/crates/tinyhivemind-openhuman/src/hosted/mod.rs +++ b/crates/tinyhivemind-openhuman/src/hosted/mod.rs @@ -6,7 +6,7 @@ //! re-expressed as configuration here. This runner asks the host for exactly //! three things, through [`EpisodeHost`]: //! -//! - **Its log**, a [`SessionLog`] over the host's own journal, which is the +//! - **Its log**, a [`SessionLog`](tinyhivemind::SessionLog) over the host's own journal, which is the //! only history there is. Each turn is seeded from it as the seat. //! - **A seat**, built by the host with the episode's tools on its belt. //! `OpenHuman` fixes a session's belt when it is built, so the host builds @@ -40,11 +40,12 @@ use std::sync::{Arc, Mutex, PoisonError}; use openhuman_core::agent::tinyagents::host::LastTurnUsage; use openhuman_core::agent::tool_policy::ToolPolicy; use openhuman_core::agent::{OpenHumanSessionHost, TurnOverrides}; -use tinyhivemind::{Conversation, Sequence, SessionLog}; +use tinyhivemind::{Conversation, Sequence}; use tinyhivemind_driver::{AgentBinding, BoundAgent}; use tinyhivemind_tools::EpisodeTools; use tinytools::Tool; +use crate::episode::Journal; use crate::raw::tools::belt_with_prefix; use crate::runner::{Lane, SeatRunner, TURN_TIMEOUT, TurnJob, unseated}; use crate::{Error, Result}; @@ -53,11 +54,8 @@ use admission::Admission; /// One hosted turn, as the host wraps it. pub type HostedTurn<'a> = Pin> + Send + 'a>>; -/// What a host gives the hosted runner. -pub trait EpisodeHost: Send + Sync + 'static { - /// The host's journal, read as a seat to seed each turn. - fn log(&self) -> &dyn SessionLog; - +/// What a host gives the hosted runner, beside the [`Journal`] it is. +pub trait EpisodeHost: Journal + 'static { /// Build the session `seat` runs on, with `belt` on it. /// /// The host builds the agent it would build anyway, adds `belt.tools` to @@ -253,7 +251,7 @@ impl SeatRunner for HostedRunner { /// Clear the seat's session, seed it from the host's log up to `since`, /// and run the brief, inside the host's wrapper. - fn turn(&self, seat: String, lane: Lane, since: Sequence, prompt: String) -> TurnJob { + fn turn(&self, seat: String, lane: Lane, since: Option, prompt: String) -> TurnJob { let host = Arc::clone(&self.host); let Some(session) = self.seats.get(&seat).map(|held| Arc::clone(&held.session)) else { return unseated(seat, lane); @@ -274,11 +272,16 @@ impl SeatRunner for HostedRunner { ..self.desk.clone() }; let window = self.window; + // This turn's usage travels with this turn: the per-seat map is the + // host's `usage(seat)` view, and a turn that ran after this one on + // the same seat may have written it before this one reads back. + let this_turn: Arc>> = Arc::new(Mutex::new(None)); Box::pin(async move { let run = { let seat = seat.clone(); let host = Arc::clone(&host); let usage = Arc::clone(&usage); + let this_turn = Arc::clone(&this_turn); async move { let history = seed::history(host.log(), conversation, &seat, since, window).await?; @@ -297,12 +300,14 @@ impl SeatRunner for HostedRunner { .map_err(Error::Harness)?; // This turn's usage, or none: a turn the session reported // nothing for must not be metered as the one before it. + let last = session.last_turn_usage(); let mut metered = usage.lock().unwrap_or_else(PoisonError::into_inner); - match session.last_turn_usage() { - Some(last) => metered.insert(seat.clone(), last), + match &last { + Some(last) => metered.insert(seat.clone(), last.clone()), None => metered.remove(&seat), }; drop(metered); + *this_turn.lock().unwrap_or_else(PoisonError::into_inner) = last; Ok(reply) } }; @@ -311,11 +316,10 @@ impl SeatRunner for HostedRunner { // reported, so a host parks what a failed turn left waiting // too. A turn that failed keeps its own error over the hook's. let outcome = host.wrap_turn(&seat, Box::pin(run)).await; - let last = usage + let last = this_turn .lock() .unwrap_or_else(PoisonError::into_inner) - .get(&seat) - .cloned(); + .take(); let finalized = host.after_turn(&seat, last.as_ref()); let result = match (outcome, finalized) { (Ok(reply), Ok(())) => Ok(reply), diff --git a/crates/tinyhivemind-openhuman/src/hosted/seed.rs b/crates/tinyhivemind-openhuman/src/hosted/seed.rs index dc8f19c0..5c07c45b 100644 --- a/crates/tinyhivemind-openhuman/src/hosted/seed.rs +++ b/crates/tinyhivemind-openhuman/src/hosted/seed.rs @@ -9,7 +9,8 @@ use tinyhivemind::{ use crate::Result; /// What `seat` was shown in `conversation` up to and including `since`, -/// newest `window` rows, as chronological `(role, content)` pairs. +/// newest `window` rows, as chronological `(role, content)` pairs; nothing +/// for a seat shown nothing yet. /// /// Projected as the seat, so a row it was not addressed on is withheld the /// same way it is everywhere else the host's log is read. The seat's own @@ -27,15 +28,20 @@ pub(super) async fn history( log: &dyn SessionLog, conversation: Conversation, seat: &str, - since: Sequence, + since: Option, window: usize, ) -> Result> { let query = SessionQuery { conversation, viewer: Viewer::Agent { id: seat.into() }, // Exclusive, so one above `since`; nothing is above the last - // sequence, so that reads unbounded rather than one short. - before: (since.0 != u64::MAX).then(|| Sequence(since.0 + 1)), + // sequence, so that reads unbounded rather than one short; and + // nothing is below the first, so a seat shown nothing reads none. + before: match since { + None => Some(Sequence(0)), + Some(Sequence(u64::MAX)) => None, + Some(since) => Some(Sequence(since.0 + 1)), + }, window, }; let rows = project_session(log, &query).await?; diff --git a/crates/tinyhivemind-openhuman/src/hosted/test.rs b/crates/tinyhivemind-openhuman/src/hosted/test.rs index b6ef5e02..be704135 100644 --- a/crates/tinyhivemind-openhuman/src/hosted/test.rs +++ b/crates/tinyhivemind-openhuman/src/hosted/test.rs @@ -16,7 +16,7 @@ use tinyhivemind_tools::{Dispatch, EpisodeTools, served_specs}; use super::EpisodeBelt; use super::seed::history; -use crate::offline::MemoryLog; +use crate::MemoryLog; fn desk(thread_root: Option) -> Conversation { Conversation { @@ -47,7 +47,7 @@ fn journal() -> MemoryLog { #[tokio::test] async fn a_seat_is_seeded_with_what_it_was_shown_its_own_rows_as_its_turns() { let log = journal(); - let seen = history(&log, desk(None), "one", Sequence(6), 30) + let seen = history(&log, desk(None), "one", Some(Sequence(6)), 30) .await .expect("reads"); assert_eq!( @@ -69,7 +69,7 @@ async fn a_seat_is_seeded_with_what_it_was_shown_its_own_rows_as_its_turns() { #[tokio::test] async fn a_row_the_seat_was_not_addressed_on_is_withheld() { let log = journal(); - let seen = history(&log, desk(None), "three", Sequence(6), 30) + let seen = history(&log, desk(None), "three", Some(Sequence(6)), 30) .await .expect("reads"); let text: Vec<&str> = seen.iter().map(|(_, content)| content.as_str()).collect(); @@ -88,11 +88,11 @@ async fn a_row_the_seat_was_not_addressed_on_is_withheld() { #[tokio::test] async fn nothing_above_the_watermark_is_seeded() { let log = journal(); - let seen = history(&log, desk(None), "one", Sequence(2), 30) + let seen = history(&log, desk(None), "one", Some(Sequence(2)), 30) .await .expect("reads"); assert_eq!(seen.len(), 2, "{seen:?}"); - let none = history(&log, desk(None), "one", Sequence(0), 30) + let none = history(&log, desk(None), "one", None, 30) .await .expect("reads"); assert!(none.is_empty(), "a first turn has no history"); @@ -101,7 +101,7 @@ async fn nothing_above_the_watermark_is_seeded() { #[tokio::test] async fn a_thread_turn_is_seeded_with_the_conversation_alone() { let log = journal(); - let seen = history(&log, desk(Some(Sequence(3))), "two", Sequence(4), 30) + let seen = history(&log, desk(Some(Sequence(3))), "two", Some(Sequence(4)), 30) .await .expect("reads"); assert_eq!( @@ -129,14 +129,14 @@ async fn the_memory_log_pages_newest_first_and_says_when_it_is_done() { Some(Sequence(3)), "a thread row names its root" ); - assert_eq!(log.latest(), Sequence(6)); + assert_eq!(log.latest(), Some(Sequence(6))); assert_eq!(log.all().len(), 6); assert_eq!(log.thread(Sequence(3)).len(), 2); assert_eq!( - log.thread_since(Sequence(3), Sequence(3)), + log.thread_since(Sequence(3), Some(Sequence(3))), vec!["@two: port 8080"] ); - let for_two = log.desk_since("two", Sequence(0)); + let for_two = log.desk_since("two", None); assert!(for_two.iter().any(|row| row.contains("which port"))); assert!(!for_two.iter().any(|row| row.contains("open work"))); assert!(format!("{log:?}").contains("engineering")); diff --git a/crates/tinyhivemind-openhuman/src/journal/README.md b/crates/tinyhivemind-openhuman/src/journal/README.md new file mode 100644 index 00000000..859818bc --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/journal/README.md @@ -0,0 +1,9 @@ +# `journal` + +`MemoryLog`: an append-only in-memory journal for one desk that is a real +`SessionLog`. Two rules decide who may read a row, and they are the rules a +host's own journal follows: a desk row with `only_for` reaches its author +and that one seat; a row in a conversation reaches the conversation's two +seats. `append` returns the sequence a row was given; `desk_since`, +`thread` and `thread_since` render rows for a reader. Always compiled: the +example, the tests and the crate's doc example are hosts over it. diff --git a/crates/tinyhivemind-openhuman/src/offline/log.rs b/crates/tinyhivemind-openhuman/src/journal/mod.rs similarity index 79% rename from crates/tinyhivemind-openhuman/src/offline/log.rs rename to crates/tinyhivemind-openhuman/src/journal/mod.rs index 1b26a926..e870a492 100644 --- a/crates/tinyhivemind-openhuman/src/offline/log.rs +++ b/crates/tinyhivemind-openhuman/src/journal/mod.rs @@ -1,8 +1,10 @@ //! An in-memory journal that is a real [`SessionLog`]. //! //! The host owns the log; this is the smallest host log that obeys the -//! port's contract, so an offline run and a test read it through exactly the -//! projection a live host's journal is read through. Two rules decide who may +//! port's contract, so an offline run, a test and a host with nothing better +//! yet read it through exactly the projection a live host's journal is read +//! through. It needs nothing the `offline` feature pulls, so it is always +//! here. Two rules decide who may //! read a row, and they are the ones a host follows too: //! //! - A desk row with `only_for` reaches its author and that one seat. @@ -33,15 +35,25 @@ pub struct Row { #[derive(Debug)] pub struct MemoryLog { desk: String, + /// The sequence the first row is given. + first: u64, rows: Mutex>, } impl MemoryLog { - /// An empty journal for `desk`. + /// An empty journal for `desk`, numbering its rows from one. #[must_use] pub fn new(desk: impl Into) -> Self { + Self::numbered_from(desk, Sequence(1)) + } + + /// An empty journal for `desk` whose first row is given `first`: a host + /// numbers its log as it likes, and some number the first row zero. + #[must_use] + pub fn numbered_from(desk: impl Into, first: Sequence) -> Self { Self { desk: desk.into(), + first: first.0, rows: Mutex::new(Vec::new()), } } @@ -59,7 +71,7 @@ impl MemoryLog { only_for: Option<&str>, ) -> Sequence { let mut rows = self.rows(); - let sequence = Sequence(rows.last().map_or(0, |row| row.sequence.0) + 1); + let sequence = Sequence(rows.last().map_or(self.first, |row| row.sequence.0 + 1)); rows.push(Row { sequence, author: author.to_owned(), @@ -70,18 +82,19 @@ impl MemoryLog { sequence } - /// The newest sequence, or zero for an empty journal. + /// The newest sequence, or `None` for an empty journal. #[must_use] - pub fn latest(&self) -> Sequence { - self.rows().last().map_or(Sequence(0), |row| row.sequence) + pub fn latest(&self) -> Option { + self.rows().last().map(|row| row.sequence) } - /// What `seat` may read on the open desk above `after`, rendered. + /// What `seat` may read on the open desk above `after`, rendered; all of + /// it for `None`. #[must_use] - pub fn desk_since(&self, seat: &str, after: Sequence) -> Vec { + pub fn desk_since(&self, seat: &str, after: Option) -> Vec { self.rows() .iter() - .filter(|row| row.sequence > after && row.thread.is_none()) + .filter(|row| after.is_none_or(|after| row.sequence > after) && row.thread.is_none()) .filter(|row| { row.only_for .as_deref() @@ -95,15 +108,15 @@ impl MemoryLog { /// in it. #[must_use] pub fn thread(&self, root: Sequence) -> Vec { - self.thread_since(root, Sequence(0)) + self.thread_since(root, None) } - /// One conversation above `after`, rendered. + /// One conversation above `after`, rendered; whole for `None`. #[must_use] - pub fn thread_since(&self, root: Sequence, after: Sequence) -> Vec { + pub fn thread_since(&self, root: Sequence, after: Option) -> Vec { self.rows() .iter() - .filter(|row| row.sequence > after) + .filter(|row| after.is_none_or(|after| row.sequence > after)) .filter(|row| row.sequence == root || row.thread == Some(root)) .map(render) .collect() diff --git a/crates/tinyhivemind-openhuman/src/lib.rs b/crates/tinyhivemind-openhuman/src/lib.rs index cae3df04..f4b424d3 100644 --- a/crates/tinyhivemind-openhuman/src/lib.rs +++ b/crates/tinyhivemind-openhuman/src/lib.rs @@ -29,6 +29,10 @@ //! words whichever runs it. The bound handle differs, which is what //! [`BoundAgent`](tinyhivemind_driver::BoundAgent) is for. //! +//! Above the runners sits [`run_episode`]: one episode from its door to +//! quiescence, over a [`Journal`] the host implements. A host builds its +//! driver, its door and a runner, and calls it. +//! //! This is the one crate in the workspace that links a harness. A host that //! seats agents some other way does not link it; it implements `BoundAgent` //! and `SeatRunner` itself. @@ -42,22 +46,36 @@ //! ```no_run //! use std::sync::Arc; //! use openhuman_core::agent::OpenHumanSessionHost; -//! use tinyhivemind::{SESSION_WINDOW, SessionLog}; +//! use tinyhivemind::{SESSION_WINDOW, Sequence, SessionLog}; +//! use tinyhivemind_driver::{Commit, Note}; +//! use tinyhivemind_openhuman::MemoryLog; //! use tinyhivemind_openhuman::{ -//! EpisodeBelt, EpisodeHost, HostedRunner, HostedTurn, Lane, LibraryHost, SeatRunner, +//! EpisodeBelt, EpisodeHost, HostedRunner, HostedTurn, Journal, Lane, LibraryHost, SeatRunner, //! }; //! use tinyhivemind_tools::{Dispatch, EpisodeTools}; //! -//! struct Desk { -//! log: L, +//! struct Desk { +//! log: MemoryLog, //! library: LibraryHost, //! } //! -//! impl EpisodeHost for Desk { +//! // The journal: a real host reads and appends its own; this one is in memory. +//! impl Journal for Desk { //! fn log(&self) -> &dyn SessionLog { //! &self.log //! } //! +//! fn commit(&self, commit: &Commit) -> tinyhivemind_openhuman::Result { +//! Ok(self.log.append(&commit.author, commit.utterance.message(), commit.thread, None)) +//! } +//! +//! fn note(&self, note: &Note) -> tinyhivemind_openhuman::Result<()> { +//! self.log.append("desk", ¬e.body, note.thread, note.only_for.as_deref()); +//! Ok(()) +//! } +//! } +//! +//! impl EpisodeHost for Desk { //! fn build_seat( //! &self, //! seat: &str, @@ -74,7 +92,7 @@ //! } //! } //! -//! # async fn run(desk: Desk) -> tinyhivemind_openhuman::Result<()> { +//! # async fn run(desk: Desk) -> tinyhivemind_openhuman::Result<()> { //! let runner = HostedRunner::seat( //! Arc::new(desk), //! Arc::new(EpisodeTools::new(["lead"])), @@ -86,7 +104,8 @@ //! runner.open("lead", Vec::new(), Dispatch { chat: "engineering".into(), parent: None }); //! // The newest row `lead` was shown before this turn: its history is read //! // from the host's log up to here, and the brief carries what is above. -//! let since = tinyhivemind::Sequence(1); +//! // `None` would be a seat shown nothing yet. +//! let since = Some(tinyhivemind::Sequence(1)); //! let (_, _, reply) = runner.turn("lead".into(), Lane::Desk, since, "Go.".into()).await; //! let events = runner.close("lead"); //! # let _ = (reply, events); @@ -95,15 +114,19 @@ //! ``` pub mod embed; +pub mod episode; pub mod error; pub mod hosted; +pub mod journal; #[cfg(any(test, feature = "offline"))] pub mod offline; pub mod raw; pub mod runner; pub use embed::{EmbedRunner, EmbedSeat}; +pub use episode::{Journal, Report, run_episode}; pub use error::{Error, Result}; pub use hosted::{EpisodeBelt, EpisodeHost, HostedRunner, HostedSeat, HostedTurn}; +pub use journal::MemoryLog; pub use raw::{LibraryHost, RawRunner, RawSeat, Route, register_seats}; pub use runner::{Lane, RunnerKind, SeatRunner, TURN_TIMEOUT, TurnJob, TurnResult}; diff --git a/crates/tinyhivemind-openhuman/src/offline/README.md b/crates/tinyhivemind-openhuman/src/offline/README.md index 9fec4486..5f459d94 100644 --- a/crates/tinyhivemind-openhuman/src/offline/README.md +++ b/crates/tinyhivemind-openhuman/src/offline/README.md @@ -7,8 +7,5 @@ the tool under, `mcp_call_tool` for an embed agent -- and a closing sentence once it sees the receipt. `Metrics` is what it saw: request bytes, and the time from a call to its receipt. `config()` is the runtime config an offline run boots with and `backend()` the stub for the core's non-inference -calls. `MemoryLog`, in `log.rs`, is an in-memory journal that is a real -`SessionLog`: a desk row with `only_for` reaches its author and that seat, -and a row in a conversation reaches the conversation's two seats, which is -the rule a host's own journal follows too. Compiled in tests and under the -`offline` feature. +calls. `MemoryLog` lives in `journal/` and is re-exported here. Compiled in +tests and under the `offline` feature. diff --git a/crates/tinyhivemind-openhuman/src/offline/mod.rs b/crates/tinyhivemind-openhuman/src/offline/mod.rs index 853bd48d..a2be2a3f 100644 --- a/crates/tinyhivemind-openhuman/src/offline/mod.rs +++ b/crates/tinyhivemind-openhuman/src/offline/mod.rs @@ -19,7 +19,6 @@ //! trip as the model experiences it, whichever road the call took. That is //! what the example's bench compares between the runners. -mod log; #[cfg(test)] mod test; @@ -27,7 +26,9 @@ use std::collections::VecDeque; use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, Instant}; -pub use log::{MemoryLog, Row}; +/// The in-memory journal, here too for a host that reaches it through the +/// feature. +pub use crate::journal::{MemoryLog, Row}; use openhuman_embed::RuntimeConfig; use serde_json::{Value, json}; use wiremock::matchers::{any, method, path}; diff --git a/crates/tinyhivemind-openhuman/src/raw/mod.rs b/crates/tinyhivemind-openhuman/src/raw/mod.rs index 0b4cc156..2cfd389d 100644 --- a/crates/tinyhivemind-openhuman/src/raw/mod.rs +++ b/crates/tinyhivemind-openhuman/src/raw/mod.rs @@ -79,13 +79,20 @@ type Contexts = Arc>>>; /// /// The loader wants `id`, `when_to_use` and a non-empty `system_prompt`; the /// prompt written here is the seat's role for a reader of the workspace, not -/// the one a session runs under. The registry is process-wide, so a host -/// seating more than one desk in one process names its seats apart. +/// the one a session runs under. +/// +/// The registry is process-wide and read **once**: the first call fixes it, +/// and a later call writes its definitions where nothing will read them and +/// fails on the first seat the fixed registry lacks. So a host registers +/// every seat it will ever run, in one call, before any session is built, +/// and a host seating more than one desk in one process names its seats +/// apart and registers them together. /// /// # Errors /// /// A seat id that is not a plain path component, the directory or a file -/// failing to write, or the registry refusing the definitions. +/// failing to write, the registry refusing the definitions, or a seat the +/// already-fixed registry does not hold. pub fn register_seats(workspace: &Path, seats: &[(&str, &str)], tools: &[String]) -> Result<()> { // A seat id names a file: one path component, and nothing a path can // be steered with. @@ -112,6 +119,8 @@ pub fn register_seats(workspace: &Path, seats: &[(&str, &str)], tools: &[String] ); std::fs::write(agents.join(format!("{id}.toml")), toml)?; } + // Set-once: a registry already read stays as it was, and the check + // below says which seat that leaves out. AgentDefinitionRegistry::init_global(workspace)?; let registry = AgentDefinitionRegistry::global().ok_or(Error::RegistryMissing)?; for (id, _) in seats { @@ -237,7 +246,7 @@ impl SeatRunner for RawRunner { /// A fresh session, seeded with what this seat has been shown and said /// so far, run once and dropped. Its belt is built for this seat and this /// turn, and every call it makes lands in the shared record. - fn turn(&self, seat: String, lane: Lane, _since: Sequence, prompt: String) -> TurnJob { + fn turn(&self, seat: String, lane: Lane, _since: Option, prompt: String) -> TurnJob { let history = self .contexts .lock() diff --git a/crates/tinyhivemind-openhuman/src/runner/mod.rs b/crates/tinyhivemind-openhuman/src/runner/mod.rs index 1fd87fdb..8e4bb251 100644 --- a/crates/tinyhivemind-openhuman/src/runner/mod.rs +++ b/crates/tinyhivemind-openhuman/src/runner/mod.rs @@ -135,9 +135,9 @@ pub trait SeatRunner: Send + Sync { /// Run one turn. The prompt is what the seat is shown this turn; `since` /// is the newest row it was shown before it, which a runner that seeds - /// from the host's log reads up to. How a seat holds context between - /// turns is the runner's business. - fn turn(&self, seat: String, lane: Lane, since: Sequence, prompt: String) -> TurnJob; + /// from the host's log reads up to, or `None` for a seat shown nothing + /// yet. How a seat holds context between turns is the runner's business. + fn turn(&self, seat: String, lane: Lane, since: Option, prompt: String) -> TurnJob; /// Open a turn: what the seat may `read`, and the chat and parent every /// call it makes must name. diff --git a/crates/tinyhivemind-openhuman/src/runner/test.rs b/crates/tinyhivemind-openhuman/src/runner/test.rs index 3f36b385..1394224a 100644 --- a/crates/tinyhivemind-openhuman/src/runner/test.rs +++ b/crates/tinyhivemind-openhuman/src/runner/test.rs @@ -15,11 +15,12 @@ use tinyhivemind_driver::standing_contract; use tinyhivemind_tools::{Dispatch, EpisodeTools, SeatEvent, served_specs}; use super::{Lane, RunnerKind, SeatRunner}; -use crate::offline::MemoryLog; +use crate::MemoryLog; use crate::{ - EmbedRunner, EpisodeBelt, EpisodeHost, HostedRunner, HostedTurn, LibraryHost, RawRunner, Route, - offline, register_seats, + EmbedRunner, EpisodeBelt, EpisodeHost, HostedRunner, HostedTurn, Journal, LibraryHost, + RawRunner, Route, offline, register_seats, }; +use tinyhivemind_driver::{Commit, Note}; /// A host with no agents of its own: its seats are library sessions, its /// log is in memory, and its wrapper is the core context a library session @@ -36,11 +37,28 @@ struct TestHost { halt: AtomicBool, } -impl EpisodeHost for TestHost { +impl Journal for TestHost { fn log(&self) -> &dyn SessionLog { &self.log } + fn commit(&self, commit: &Commit) -> crate::Result { + Ok(self.log.append( + &commit.author, + commit.utterance.message(), + commit.thread, + commit.only_for.as_deref(), + )) + } + + fn note(&self, note: &Note) -> crate::Result<()> { + self.log + .append("desk", ¬e.body, note.thread, note.only_for.as_deref()); + Ok(()) + } +} + +impl EpisodeHost for TestHost { fn build_seat(&self, seat: &str, belt: EpisodeBelt) -> crate::Result { let policy = belt.admit(None); self.library.session(seat, &self.prompt, belt.tools, policy) @@ -108,7 +126,7 @@ fn each_runner_states_its_own_mechanics_and_nothing_else() { } /// One turn through the seam: open, run, close. -async fn one_turn(runner: &R, since: Sequence) -> (String, Vec) { +async fn one_turn(runner: &R, since: Option) -> (String, Vec) { let bindings = runner.bindings(); assert_eq!(bindings.len(), 1); assert_eq!(bindings[0].hive_agent_id, "lead"); @@ -140,11 +158,28 @@ struct PlainHost { library: LibraryHost, } -impl EpisodeHost for PlainHost { +impl Journal for PlainHost { fn log(&self) -> &dyn SessionLog { &self.log } + fn commit(&self, commit: &Commit) -> crate::Result { + Ok(self.log.append( + &commit.author, + commit.utterance.message(), + commit.thread, + commit.only_for.as_deref(), + )) + } + + fn note(&self, note: &Note) -> crate::Result<()> { + self.log + .append("desk", ¬e.body, note.thread, note.only_for.as_deref()); + Ok(()) + } +} + +impl EpisodeHost for PlainHost { fn build_seat(&self, seat: &str, belt: EpisodeBelt) -> crate::Result { let policy = belt.admit(None); self.library @@ -155,7 +190,7 @@ impl EpisodeHost for PlainHost { /// A hosted seat on a host that keeps every default, run once on the desk /// and once in a thread it is not in: the defaults hold, the thread turn is /// seeded from the thread, and a call outside its thread is refused. -async fn plain(library: LibraryHost, contract: &str) { +async fn plain(library: LibraryHost) { let log = MemoryLog::new("engineering"); log.append("operator", "state the root cause", None, None); let host = Arc::new(PlainHost { log, library }); @@ -168,7 +203,6 @@ async fn plain(library: LibraryHost, contract: &str) { SESSION_WINDOW, ) .expect("hosted seats"); - let _ = contract; let (reply, events) = one_turn(&runner, host.log.latest()).await; assert!(!reply.is_empty()); assert_eq!( @@ -188,7 +222,7 @@ async fn plain(library: LibraryHost, contract: &str) { .turn( "lead".into(), Lane::Thread(Sequence(1)), - Sequence(1), + Some(Sequence(1)), "In the thread.".into(), ) .await; @@ -250,7 +284,7 @@ fn one_completion(name: &str, events: &[SeatEvent]) { async fn again(raw: &RawRunner, host: &TestHost, hosted: &HostedRunner) { // A second raw turn is seeded with the first: what the seat said is what // it is shown, and the record starts empty again. - let (_, again) = one_turn(raw, Sequence(0)).await; + let (_, again) = one_turn(raw, None).await; assert_eq!(again.len(), 1); host.log.append("lead", "COMPLETE: done", None, None); let (_, again) = one_turn(hosted, host.log.latest()).await; @@ -263,13 +297,12 @@ async fn again(raw: &RawRunner, host: &TestHost, hosted: &HostedRunner /// rather than seated as a ghost. async fn ghosts(embed: &EmbedRunner, raw: &RawRunner, hosted: &HostedRunner) { for outcome in [ - raw.turn("ghost".into(), Lane::Desk, Sequence(0), "?".into()) - .await, + raw.turn("ghost".into(), Lane::Desk, None, "?".into()).await, hosted - .turn("ghost".into(), Lane::Desk, Sequence(0), "?".into()) + .turn("ghost".into(), Lane::Desk, None, "?".into()) .await, embed - .turn("ghost".into(), Lane::Desk, Sequence(0), "?".into()) + .turn("ghost".into(), Lane::Desk, None, "?".into()) .await, ] { assert!( @@ -331,7 +364,7 @@ async fn halts(host: &TestHost, hosted: &HostedRunner) { .turn( "lead".into(), Lane::Thread(Sequence(u64::MAX)), - Sequence(u64::MAX), + Some(Sequence(u64::MAX)), "Once more.".into(), ) .await; @@ -458,8 +491,8 @@ async fn both_runners() { .expect("the library boots"); let (host, hosted) = hosted(library, &contract(RunnerKind::Hosted)); - let (embed_reply, embed_events) = one_turn(&embed, Sequence(0)).await; - let (raw_reply, raw_events) = one_turn(&raw, Sequence(0)).await; + let (embed_reply, embed_events) = one_turn(&embed, None).await; + let (raw_reply, raw_events) = one_turn(&raw, None).await; // Seeded from the host's log: the operator's row is history, not brief. let (hosted_reply, hosted_events) = one_turn(&hosted, host.log.latest()).await; assert_eq!( @@ -488,7 +521,7 @@ async fn both_runners() { again(&raw, &host, &hosted).await; ghosts(&embed, &raw, &hosted).await; - plain(host.library.clone(), &contract(RunnerKind::Hosted)).await; + plain(host.library.clone()).await; halts(&host, &hosted).await; metrics.reset(); assert_eq!(metrics.snapshot().requests, 0); diff --git a/examples/openhuman/README.md b/examples/openhuman/README.md index ac89d2c5..04c65ffa 100644 --- a/examples/openhuman/README.md +++ b/examples/openhuman/README.md @@ -48,18 +48,18 @@ corpus and paid campaign described in | `src/bin/pe1006_hive.rs` | OpenRouter GPT-OSS completion-driven hive with stable OpenHuman sessions and live TypeSafe routing. | | `src/bin/deepswe_hive.rs` | Hermetic four-seat software-engineering hive over a caller-prepared disposable Git checkout. | | `src/bin/conducted.rs` | A live completion-driven episode: the loop stepped through the `Conductor`, any of the adapter's three runners, a hidden-profile desk of five seats over OpenRouter with live Jev routing, or offline against the adapter's scripted model. `CONDUCTED_DESK=login` (default) diagnoses a regression; `CONDUCTED_DESK=triage` hands off three tickets on a budget of two, to fire the budget, the broadcast that completes its author, and the in-thread `ask` refusal. | -| `src/bin/conducted/hosted.rs` | This example as an `EpisodeHost`: its journal is the log, a seat is a library session with the episode's belt, and the wrapper is the core context. The runners themselves live in `tinyhivemind-openhuman`. | +| `src/bin/conducted/hosted.rs` | This example as a host: `DeskJournal`, its in-memory log with the prompt and the log lines, for every runner; and `DeskHost`, an `EpisodeHost` whose seats are library sessions with the episode's belt. The runners and the loop live in `tinyhivemind-openhuman`. | | `src/bin/conducted/jev.rs` | The live `SystemOneTransport` over `tinyjevclient`, bridged through the wire form. | | `deepswe-sandbox/` | Reproducible local Docker image used for agent shell and test execution. | -## `conducted`: one loop, two runners +## `conducted`: one loop, three runners -`src/bin/conducted.rs` steps one completion-driven episode the way a host steps -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`. +`src/bin/conducted.rs` runs one completion-driven episode through +`tinyhivemind_openhuman::run_episode`: it builds the hive, the driver, the +door and a runner, and implements `Journal` over an in-memory log. The +journal, the prompt and the log lines are the host's; the wave loop is the +adapter's, and 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 three implementations the loop cannot tell apart: diff --git a/examples/openhuman/src/bin/conducted.rs b/examples/openhuman/src/bin/conducted.rs index 90561f79..f97273ec 100644 --- a/examples/openhuman/src/bin/conducted.rs +++ b/examples/openhuman/src/bin/conducted.rs @@ -35,34 +35,28 @@ use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -use conducted::hosted::DeskHost; +use conducted::hosted::{DESK_PREAMBLE, DeskHost, DeskJournal}; use conducted::jev::LiveJev; use openhuman_embed::{Access, Provider, Runtime, RuntimeConfig, Workspace}; use tinyhivemind::SESSION_WINDOW; use tinyhivemind::desk::{Desk, ResponderMode}; use tinyhivemind::responder::Probability; -use tinyhivemind::speech::Utterance; use tinyhivemind_driver::{ - BoundHive, BroadcastRouting, CompletionDriver, ConductPolicy, Conductor, Door, Event, - HiveGraph, Refusal, Step, standing_contract, + BoundHive, BroadcastRouting, CompletionDriver, ConductPolicy, Door, HiveGraph, + standing_contract, }; use tinyhivemind_embed::{ ConversationKind, ConversationRef, RouteCandidate, Router, RouterFuture, RoutingPlan, RoutingPolicy, RoutingRequest, RoutingSource, route_message, }; -use tinyhivemind_openhuman::offline::MemoryLog; use tinyhivemind_openhuman::{ - EmbedRunner, HostedRunner, Lane, LibraryHost, RawRunner, Route, RunnerKind, SeatRunner, - TurnJob, offline, + EmbedRunner, HostedRunner, Journal, LibraryHost, MemoryLog, RawRunner, Route, RunnerKind, + SeatRunner, offline, run_episode, }; -use tinyhivemind_tools::{Dispatch, EpisodeTools}; +use tinyhivemind_tools::EpisodeTools; use tinyhivemind_typesafe::JevRouter; -/// What the host says about the desk, before the episode's own contract. -const DESK_PREAMBLE: &str = "\ -You are one seat on a desk. You have no codebase, shell or filesystem -- only -the desk's messages and your own judgement. Never ask for permission and never -wait to be told to continue; nobody will answer."; + /// A desk: who sits at it, what each seat privately knows, and the task. struct Scenario { @@ -197,8 +191,6 @@ production snapshot taken after the deploy.", ], }; -/// How much of a reply that recorded nothing is shown in the log. -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; @@ -322,21 +314,22 @@ async fn run() -> anyhow::Result<()> { None => { println!("[runner] {}", kind.name()); let journal = Arc::new(MemoryLog::new(desk_id)); + let desk = host.journal(&journal, false); let report = match kind { RunnerKind::Embed => { let runtime = host.runtime().await?; let runner = host.embed(&runtime, 0).await?; - episode(runner, host.setup(kind, false), journal).await? + episode(runner, host.setup(kind, false), journal, &desk).await? } RunnerKind::Raw => { host.prepare_raw()?; let runner = host.raw(0).await?; - episode(runner, host.setup(kind, false), journal).await? + episode(runner, host.setup(kind, false), journal, &desk).await? } RunnerKind::Hosted => { host.prepare_raw()?; - let runner = host.hosted(&journal).await?; - episode(runner, host.setup(kind, false), journal).await? + let (runner, desk) = host.hosted(&journal, false).await?; + episode(runner, host.setup(kind, false), journal, &*desk).await? } }; let _ = report; @@ -381,7 +374,6 @@ impl Host { kind, ids: self.ids.clone(), candidates: self.candidates.clone(), - briefs: self.briefs.clone(), live: self.live, quiet, } @@ -451,10 +443,20 @@ impl Host { Ok(runner) } + /// This desk as a journal over `journal`, for every runner. + fn journal(&self, journal: &Arc, quiet: bool) -> DeskJournal { + DeskJournal::new(Arc::clone(journal), self.briefs.clone(), quiet) + } + /// Seat the hosted runner over `journal`, which is also the log the - /// episode appends to. Its seats are registered by `prepare_raw`, the - /// same definitions the raw seats resolve. - async fn hosted(&self, journal: &Arc) -> anyhow::Result> { + /// episode appends to, and return the host it was seated on, which is + /// the episode's journal too. Its seats are registered by + /// `prepare_raw`, the same definitions the raw seats resolve. + async fn hosted( + &self, + journal: &Arc, + quiet: bool, + ) -> anyhow::Result<(HostedRunner, Arc)> { let library = LibraryHost::boot( &self.config, &self.backend_url, @@ -468,15 +470,16 @@ impl Host { .iter() .map(|(id, brief)| (id.clone(), format!("{brief}\n\n{contract}"))) .collect(); - let host = Arc::new(DeskHost::new(Arc::clone(journal), library, prompts)); - Ok(HostedRunner::seat( - host, + let host = Arc::new(DeskHost::new(self.journal(journal, quiet), library, prompts)); + let runner = HostedRunner::seat( + Arc::clone(&host), Arc::new(EpisodeTools::new(self.ids.iter().cloned())), &self.ids, self.scenario.id, self.scenario.name, SESSION_WINDOW, - )?) + )?; + Ok((runner, host)) } } @@ -519,18 +522,19 @@ async fn bench_runners( let runtime = &runtime; let run = move |index: u32| async move { let journal = Arc::new(MemoryLog::new(host.scenario.id)); + let desk = host.journal(&journal, true); match kind { RunnerKind::Embed => { let runner = host.embed(runtime, index).await?; - episode(runner, host.setup(kind, true), journal).await + episode(runner, host.setup(kind, true), journal, &desk).await } RunnerKind::Raw => { let runner = host.raw(index).await?; - episode(runner, host.setup(kind, true), journal).await + episode(runner, host.setup(kind, true), journal, &desk).await } RunnerKind::Hosted => { - let runner = host.hosted(&journal).await?; - episode(runner, host.setup(kind, true), journal).await + let (runner, desk) = host.hosted(&journal, true).await?; + episode(runner, host.setup(kind, true), journal, &*desk).await } } }; @@ -615,7 +619,6 @@ struct Setup { kind: RunnerKind, ids: Vec, candidates: Vec, - briefs: BTreeMap, live: bool, /// Skip the journal dump at the end: a bench prints one table instead. quiet: bool, @@ -636,17 +639,17 @@ struct Report { /// 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( +async fn episode( runner: R, setup: Setup, journal: Arc, + desk: &J, ) -> anyhow::Result { let Setup { scenario, kind, ids, candidates, - briefs, live, quiet, } = setup; @@ -724,7 +727,9 @@ async fn episode( roster_version: 1, thread_context: &[], }; - let mut conductor = Conductor::open( + let outcome = run_episode( + desk, + &runner, &driver, routing, ConductPolicy::default(), @@ -735,120 +740,22 @@ async fn episode( starters, opened_at, }, - )?; - - let settled: anyhow::Result<()> = 'episode: loop { - if conductor.finished() { - break Ok(()); - } - for step in conductor.begin_wave() { - take(&journal, step); - } - let turns = match conductor.turns() { - Ok(turns) => turns, - Err(error) => break Err(error.into()), - }; - let mut jobs: Vec = Vec::new(); - 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( - &turn.seat, - match turn.thread() { - None => rows.clone(), - Some(root) => journal.thread(root), - }, - Dispatch { - chat: desk_id.into(), - parent: turn.thread().map(|root| root.0.to_string()), - }, - ); - 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[&turn.seat], - brief.render() - ); - let lane = turn.thread().map_or(Lane::Desk, Lane::Thread); - jobs.push(runner.turn(turn.seat.clone(), lane, turn.since, prompt)); - } - let outcomes = futures::future::join_all(jobs).await; - for (seat_id, lane, outcome) in outcomes { - let where_ = match lane { - Lane::Desk => String::new(), - Lane::Thread(root) => format!(" in thread {}", root.0), - }; - match &outcome { - Some(Ok(reply)) => eprintln!( - "[turn] @{seat_id}{where_} replied ({} chars)", - reply.chars().count() - ), - Some(Err(error)) => eprintln!("[turn] @{seat_id}{where_} failed: {error}"), - None => eprintln!("[turn] @{seat_id}{where_} timed out"), - } - // Close the turn first: the record refuses a call on a closed - // turn, so nothing can land after this point is read. - let events = runner.close(&seat_id); - for refused in runner.tools().drain_refusals(&seat_id) { - eprintln!( - "[refused] @{seat_id}{where_} `{}`: {}", - refused.tool, refused.reason - ); - } - if events.is_empty() { - eprintln!("[no tool call] @{seat_id}{where_}"); - // What the seat wrote instead: the only trace of a refusal - // it read, or of a deliverable it typed rather than recorded. - if let Some(Ok(reply)) = &outcome { - let shown: String = reply.chars().take(REPLY_SHOWN).collect(); - let cut = if reply.chars().count() > REPLY_SHOWN { - " [...]" - } else { - "" - }; - eprintln!(" {}{cut}", shown.replace('\n', "\n ")); - } - } - 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)); - } - 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()); - } - } - Ok(Some(step)) => take(&journal, step), - Err(error) => break 'episode Err(error.into()), - } - } + ) + .await; + let report = match &outcome { + Ok(report) => *report, + Err(_) => tinyhivemind_openhuman::Report::default(), }; - println!( "turns {} | routes {} | waves {} | discharged {} | settled {} | conversations {}", - conductor.turns_run(), + report.turns, router .as_ref() .map_or(0, |r| r.calls.load(Ordering::SeqCst)), - conductor.waves(), - conductor.discharged(), - conductor.state().episode().settled(), - conductor.conversations() + report.waves, + report.discharged, + report.settled, + report.conversations ); for row in journal.all().iter().filter(|_| !quiet) { let scope = match (row.thread, row.only_for.as_deref()) { @@ -861,6 +768,7 @@ async fn episode( row.sequence.0, row.author, row.body ); } + let settled: anyhow::Result<()> = outcome.map(|_| ()).map_err(Into::into); settled?; // Offline, the run is a proof and says so: the scripted seat's tool call // must have become a desk row -- natively through the belt, or over the @@ -883,96 +791,14 @@ async fn episode( } } let report = Report { - turns: conductor.turns_run(), - waves: conductor.waves(), + turns: report.turns, + waves: report.waves, wall: started.elapsed(), }; - drop(conductor); drop(runner); Ok(report) } -/// A note the desk says, appended; an event, logged. -fn take(journal: &MemoryLog, step: Step) { - match step { - Step::Note(note) => { - journal.append("desk", ¬e.body, note.thread, note.only_for.as_deref()); - } - 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 - ), - 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. -fn describe(utterance: &Utterance) -> String { - match utterance { - Utterance::Post { message } => message.clone(), - Utterance::Broadcast { message } => format!("BROADCAST: {message}"), - Utterance::Ask { to, message } => format!("asks @{to}: {message}"), - Utterance::Dm { message, .. } => message.clone(), - Utterance::CompleteEpisode { message } => format!("COMPLETE: {message}"), - } -} - /// Nothing running but the one subsystem the episode's tools arrive through. fn required(name: &str) -> anyhow::Result { std::env::var(name).map_err(|_| anyhow::anyhow!("{name} must be set for a live run")) diff --git a/examples/openhuman/src/bin/conducted/hosted.rs b/examples/openhuman/src/bin/conducted/hosted.rs index 9ba06a26..c3010cad 100644 --- a/examples/openhuman/src/bin/conducted/hosted.rs +++ b/examples/openhuman/src/bin/conducted/hosted.rs @@ -1,22 +1,197 @@ -//! The host the hosted runner asks for, as this example is one. +//! This example as a host: its journal, and its seats. //! -//! A real host -- OpenCompany -- answers these three with its journal, the -//! agents it already builds, and the task-locals its tools read. This one -//! has no agents of its own, so a seat is a library session carrying only -//! the episode's tools, and the wrapper is the core context such a session -//! runs under. The log is the same journal the episode loop appends to. +//! A real host -- OpenCompany -- answers these with its journal, the agents +//! it already builds, and the task-locals its tools read. This one has no +//! agents of its own, so a seat is a library session carrying only the +//! episode's tools, and the wrapper is the core context such a session runs +//! under. The log is the same in-memory journal the episode appends to. use std::collections::BTreeMap; use std::sync::Arc; use openhuman_core::agent::OpenHumanSessionHost; -use tinyhivemind::SessionLog; -use tinyhivemind_openhuman::offline::MemoryLog; -use tinyhivemind_openhuman::{EpisodeBelt, EpisodeHost, HostedTurn, LibraryHost}; +use tinyhivemind::speech::Utterance; +use tinyhivemind::{Sequence, SessionLog}; +use tinyhivemind_driver::{Commit, EpisodeBrief, Event, Note, Refusal}; +use tinyhivemind_openhuman::{ + EpisodeBelt, EpisodeHost, HostedTurn, Journal, Lane, LibraryHost, MemoryLog, TurnResult, +}; -/// This example's desk, as a host. -pub struct DeskHost { +/// What the host says about the desk, before the episode's own contract. +pub const DESK_PREAMBLE: &str = "\ +You are one seat on a desk. You have no codebase, shell or filesystem -- only +the desk's messages and your own judgement. Never ask for permission and never +wait to be told to continue; nobody will answer."; + +/// How much of a reply that recorded nothing is shown in the log. +const REPLY_SHOWN: usize = 600; + +/// This example's journal, with the prompt and the log lines a host owns. +pub struct DeskJournal { log: Arc, + /// Each seat's brief: who it is, and what it privately knows. + briefs: BTreeMap, + /// Print nothing per turn: a bench prints one table instead. + quiet: bool, +} + +impl DeskJournal { + pub fn new(log: Arc, briefs: BTreeMap, quiet: bool) -> Self { + Self { log, briefs, quiet } + } +} + +/// How a row reads on the desk. +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}"), + } +} + +impl Journal for DeskJournal { + fn log(&self) -> &dyn SessionLog { + &*self.log + } + + fn commit(&self, commit: &Commit) -> tinyhivemind_openhuman::Result { + Ok(self.log.append( + &commit.author, + &describe(&commit.utterance), + commit.thread, + commit.only_for.as_deref(), + )) + } + + fn note(&self, note: &Note) -> tinyhivemind_openhuman::Result<()> { + self.log + .append("desk", ¬e.body, note.thread, note.only_for.as_deref()); + Ok(()) + } + + /// What the host owns first; what the episode knows after. + fn compose(&self, seat: &str, brief: &EpisodeBrief) -> String { + format!( + "## The desk\n{DESK_PREAMBLE}\n\n## Who you are\n{}\n\n{}", + self.briefs[seat], + brief.render() + ) + } + + fn event(&self, 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 + ), + 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 { + "" + } + ), + } + } + + fn turn_done( + &self, + seat: &str, + lane: Lane, + outcome: &TurnResult, + refused: &[tinyhivemind_tools::Refusal], + recorded: usize, + ) { + if self.quiet { + return; + } + let where_ = match lane { + Lane::Desk => String::new(), + Lane::Thread(root) => format!(" in thread {}", root.0), + }; + match outcome { + Some(Ok(reply)) => eprintln!( + "[turn] @{seat}{where_} replied ({} chars)", + reply.chars().count() + ), + Some(Err(error)) => eprintln!("[turn] @{seat}{where_} failed: {error}"), + None => eprintln!("[turn] @{seat}{where_} timed out"), + } + for refusal in refused { + eprintln!( + "[refused] @{seat}{where_} `{}`: {}", + refusal.tool, refusal.reason + ); + } + if recorded == 0 { + // What the seat wrote instead, marked as what it is: not a desk + // row, and the only trace of a refusal it read or of a + // deliverable it typed rather than recorded. + eprintln!("[no tool call] @{seat}{where_} -- reply discarded, not recorded:"); + if let Some(Ok(reply)) = outcome { + let shown: String = reply.chars().take(REPLY_SHOWN).collect(); + let cut = if reply.chars().count() > REPLY_SHOWN { + " [...]" + } else { + "" + }; + eprintln!(" {}{cut}", shown.replace('\n', "\n ")); + } + } + } +} + +/// This example's desk, as a host of its own seats. +pub struct DeskHost { + journal: DeskJournal, library: LibraryHost, /// Each seat's standing prompt: its brief and the contract. prompts: BTreeMap, @@ -24,23 +199,53 @@ pub struct DeskHost { impl DeskHost { pub fn new( - log: Arc, + journal: DeskJournal, library: LibraryHost, prompts: BTreeMap, ) -> Self { Self { - log, + journal, library, prompts, } } } -impl EpisodeHost for DeskHost { +impl Journal for DeskHost { fn log(&self) -> &dyn SessionLog { - &*self.log + self.journal.log() + } + + fn commit(&self, commit: &Commit) -> tinyhivemind_openhuman::Result { + self.journal.commit(commit) } + fn note(&self, note: &Note) -> tinyhivemind_openhuman::Result<()> { + self.journal.note(note) + } + + fn compose(&self, seat: &str, brief: &EpisodeBrief) -> String { + self.journal.compose(seat, brief) + } + + fn event(&self, event: &Event) { + self.journal.event(event); + } + + fn turn_done( + &self, + seat: &str, + lane: Lane, + outcome: &TurnResult, + refused: &[tinyhivemind_tools::Refusal], + recorded: usize, + ) { + self.journal + .turn_done(seat, lane, outcome, refused, recorded); + } +} + +impl EpisodeHost for DeskHost { fn build_seat( &self, seat: &str,