From 303ecf162133431473aa3952957235d572a0270e Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 23 Sep 2026 02:26:49 +0530 Subject: [PATCH 1/3] Park a seat on the host, and carry its other conversations in the brief A turn that stops on something only the host can settle -- an approval -- read as a silent one: nudged, then a stall. `record_parked` holds the seat where it stopped, not nudged, not stalled, not proposed, with a parked askee's conversation waiting rather than concluding; `resume_seat` owes it a turn there. Nothing due with a seat parked is a wait, and `Error::Parked` is the host giving up on one. `EpisodeBrief::elsewhere` carries the newest rows of the seat's other conversations, rendered as context after the turn's own rows. Co-Authored-By: Claude Fable 5.1 --- .../tinyhivemind-driver/src/conduct/README.md | 8 + crates/tinyhivemind-driver/src/conduct/mod.rs | 61 ++++++- .../tinyhivemind-driver/src/conduct/steps.rs | 16 ++ .../src/conduct/test/mod.rs | 1 + .../src/conduct/test/parked.rs | 161 ++++++++++++++++++ .../src/conduct/test/support.rs | 15 ++ .../src/conduct/test/wire.rs | 8 + .../tinyhivemind-driver/src/conduct/wave.rs | 2 +- .../tinyhivemind-driver/src/driver/brief.rs | 53 +++++- .../src/driver/brief/test.rs | 60 ++++++- crates/tinyhivemind-driver/src/driver/mod.rs | 2 +- crates/tinyhivemind-driver/src/error/mod.rs | 7 + crates/tinyhivemind-driver/src/lib.rs | 4 +- 13 files changed, 388 insertions(+), 10 deletions(-) create mode 100644 crates/tinyhivemind-driver/src/conduct/test/parked.rs diff --git a/crates/tinyhivemind-driver/src/conduct/README.md b/crates/tinyhivemind-driver/src/conduct/README.md index a60a803d..c9db27c3 100644 --- a/crates/tinyhivemind-driver/src/conduct/README.md +++ b/crates/tinyhivemind-driver/src/conduct/README.md @@ -23,6 +23,14 @@ The rules, each with the decision it comes from: has been shown everything is told once per assignment and owed a turn. A seat asked that took its turn without answering is told once and owed a turn; a second silence stands. +- **Parking**: a turn that stopped on something only the host can settle + -- an approval, typically -- is recorded with `record_parked` instead of + its calls. The seat is held where it parked: not nudged for silence, not + counted toward a stall, not proposed again, and a parked askee's + conversation waits with it rather than concluding for want of a turn. + Nothing due with a seat parked is a wait, not a stall; `parked` says who, + and `resume_seat` puts the seat back in the next wave, owed a turn where + it parked. `Event::Parked` and `Event::Resumed` mark both. - **Sorting**: a broadcast or an ask made inside a conversation is desk work; only a post or a completion is a row of the conversation. - **Refusals**: a completion the ledger refuses is explained to the seat on diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs index 5e649064..0721c510 100644 --- a/crates/tinyhivemind-driver/src/conduct/mod.rs +++ b/crates/tinyhivemind-driver/src/conduct/mod.rs @@ -124,6 +124,9 @@ pub struct Conductor<'a, A: BoundAgent> { shown: BTreeMap, /// The assignment each seat was last nudged for on the desk. desk_nudged: BTreeMap, + /// Seats held on the host, by the thread they parked in (`None` for the + /// desk): not nudged, not stalled, not proposed, until released. + parked: BTreeMap>, turns: u64, waves: u64, discharged: u64, @@ -200,6 +203,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { concluded: Vec::new(), shown: BTreeMap::new(), desk_nudged: BTreeMap::new(), + parked: BTreeMap::new(), turns: 0, waves: 0, discharged: 0, @@ -255,6 +259,10 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { self.waves += 1; let mut steps = Vec::new(); for seat in self.state.stalled() { + // A parked seat is waiting on the host, not on the desk. + if self.parked.contains_key(&seat) { + continue; + } let assigned_at = self .state .episode() @@ -304,7 +312,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { let mut turns = Vec::new(); for child in self.children.values() { for seat in self.pending(&child.state)? { - if !taken.insert(seat.clone()) { + if self.parked.contains_key(&seat) || !taken.insert(seat.clone()) { continue; } // The ask row is the first thing a seat is shown in the @@ -329,7 +337,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { } } for seat in self.pending(&self.state)? { - if !taken.insert(seat.clone()) { + if self.parked.contains_key(&seat) || !taken.insert(seat.clone()) { continue; } let since = self.state.seen().delivered_through.get(&seat).copied(); @@ -339,15 +347,60 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { since, }); } - if turns.is_empty() && self.children.is_empty() { + if turns.is_empty() && self.children.is_empty() && self.parked.is_empty() { return Err(Error::Stalled { seats: self.state.stalled(), }); } - self.wave.begin(turns.is_empty()); + // Nothing due with a seat parked is a wait, not an end: no + // conversation concludes for it. + self.wave.begin(turns.is_empty() && self.parked.is_empty()); Ok(turns) } + /// The seats held on the host, in seat order. An empty wave with any of + /// these is the host's to end: it releases one with + /// [`resume_seat`](Self::resume_seat), or gives up. + #[must_use] + pub fn parked(&self) -> Vec { + self.parked.keys().cloned().collect() + } + + /// A turn stopped on something only the host can settle -- an approval, + /// typically -- and recorded nothing. The seat is held where it parked: + /// it is not nudged for silence, does not stall the episode, and is not + /// proposed again until the host releases it. A parked askee's + /// conversation waits with it. + pub fn record_parked(&mut self, turn: &Turn) { + self.turns += 1; + let thread = turn.thread(); + if let Some(child) = thread.and_then(|root| self.children.get_mut(&root)) { + // Not a silence: the askee did not take this turn, for nudging. + child.turned = false; + } + self.parked.insert(turn.seat.clone(), thread); + self.wave.event(Event::Parked { + seat: turn.seat.clone(), + thread, + }); + } + + /// The host settled what a seat parked on: it is owed a turn where it + /// parked, in the next wave. A seat that is not parked is left as it is. + pub fn resume_seat(&mut self, seat: &str) { + let Some(thread) = self.parked.remove(seat) else { + return; + }; + match thread.and_then(|root| self.children.get_mut(&root)) { + Some(child) => child.state.owe_turn(seat), + None => self.state.owe_turn(seat), + } + self.wave.event(Event::Resumed { + seat: seat.to_owned(), + thread, + }); + } + fn pending(&self, state: &DriverState) -> Result> { Ok(self .driver diff --git a/crates/tinyhivemind-driver/src/conduct/steps.rs b/crates/tinyhivemind-driver/src/conduct/steps.rs index 63ee9c67..d6211a91 100644 --- a/crates/tinyhivemind-driver/src/conduct/steps.rs +++ b/crates/tinyhivemind-driver/src/conduct/steps.rs @@ -134,6 +134,22 @@ pub enum Event { /// The thread, or `None` on the desk. thread: Option, }, + /// A seat's turn stopped on something only the host can settle, and the + /// seat is held: not nudged, not stalled, not proposed, until the host + /// releases it. + Parked { + /// The seat. + seat: String, + /// The thread, or `None` on the desk. + thread: Option, + }, + /// The host released a parked seat: it is owed a turn where it parked. + Resumed { + /// The seat. + seat: String, + /// The thread, or `None` on the desk. + thread: Option, + }, /// A broadcast was placed. Broadcast { /// The author. diff --git a/crates/tinyhivemind-driver/src/conduct/test/mod.rs b/crates/tinyhivemind-driver/src/conduct/test/mod.rs index 93041eec..6c08661d 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/mod.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/mod.rs @@ -5,5 +5,6 @@ mod conversations; mod desk; mod door; mod links; +mod parked; mod support; mod wire; diff --git a/crates/tinyhivemind-driver/src/conduct/test/parked.rs b/crates/tinyhivemind-driver/src/conduct/test/parked.rs new file mode 100644 index 00000000..efaf3e53 --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/parked.rs @@ -0,0 +1,161 @@ +//! A parked seat: held on the host, not nudged, not stalled, not proposed, +//! and back where it stopped once released. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use super::support::{Journal, ask, complete, hive, policy, seats, two_seat, wave, wave_parking}; +use crate::conduct::{ConductPolicy, Conductor, Event}; +use crate::driver::BroadcastRouting; +use crate::{CompletionDriver, Error}; +use tinyhivemind::Sequence; + +#[test] +fn a_parked_desk_seat_is_held_until_the_host_releases_it() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + // One's turn stops on the host: an approval, say. + let parked = wave_parking(&mut conductor, &journal, &[], &["one"]).expect("wave"); + assert_eq!(seats(&parked.turns), vec![("one", None)]); + assert!(matches!( + parked.events.as_slice(), + [Event::Parked { seat, thread: None }] if seat == "one" + )); + assert_eq!(conductor.parked(), vec!["one".to_owned()]); + assert_eq!(conductor.turns_run(), 1, "a parked turn ran"); + assert!(!conductor.finished()); + // The next wave nudges nobody, proposes nothing, and is not a stall: + // the seat is waiting on the host, and the host is asked. + let waiting = wave(&mut conductor, &journal, &[]).expect("not a stall"); + assert!(waiting.turns.is_empty()); + assert!(waiting.events.is_empty(), "{:?}", waiting.events); + assert!( + !journal + .bodies() + .iter() + .any(|body| body.contains("open work")), + "a parked seat is not told it is silent" + ); + // Released, it is owed its turn where it parked, and completes. + conductor.resume_seat("one"); + let resumed = wave( + &mut conductor, + &journal, + &[("one", vec![complete("approved and done")])], + ) + .expect("wave"); + assert_eq!(seats(&resumed.turns), vec![("one", None)]); + assert!(matches!( + resumed.events.as_slice(), + [Event::Resumed { seat, thread: None }] if seat == "one" + )); + assert!(conductor.parked().is_empty()); + assert!(conductor.finished()); +} + +#[test] +fn a_parked_askee_holds_its_conversation_open_and_answers_once_released() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + wave( + &mut conductor, + &journal, + &[("one", vec![ask("two", "may I ship?")])], + ) + .expect("wave"); + let root = Sequence(2); + // Two parks in the thread; one, woken by its own ask, says nothing. + let parked = wave_parking(&mut conductor, &journal, &[], &["two"]).expect("wave"); + assert_eq!( + seats(&parked.turns), + vec![("two", Some(root)), ("one", None)] + ); + assert!( + parked + .events + .iter() + .any(|event| matches!(event, Event::Parked { seat, thread: Some(at) } if seat == "two" && *at == root)) + ); + assert!( + !parked + .events + .iter() + .any(|event| matches!(event, Event::Nudged { seat, .. } if seat == "two")), + "a parked askee is not a silent one: {:?}", + parked.events + ); + assert_eq!( + conductor.conversations(), + 0, + "the conversation waits with it" + ); + // The asker, silent on the desk, is nudged as ever; the parked askee is + // not proposed, and the conversation is not concluded for want of it. + let waiting = wave(&mut conductor, &journal, &[]).expect("not a stall"); + assert_eq!(seats(&waiting.turns), vec![("one", None)]); + assert!( + waiting + .events + .iter() + .all(|event| matches!(event, Event::Nudged { seat, thread: None } if seat == "one")), + "{:?}", + waiting.events + ); + assert_eq!(conductor.conversations(), 0); + conductor.resume_seat("two"); + let answered = wave( + &mut conductor, + &journal, + &[("two", vec![complete("ship it")])], + ) + .expect("wave"); + assert_eq!(seats(&answered.turns)[0], ("two", Some(root))); + assert_eq!( + conductor.conversations(), + 1, + "answered, the conversation concluded" + ); +} + +#[test] +fn releasing_a_seat_that_is_not_parked_changes_nothing_and_a_stall_is_still_a_stall() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor: Conductor<'_, _> = + two_seat(&driver, routing, ConductPolicy::default(), &journal); + conductor.resume_seat("one"); + let first = wave(&mut conductor, &journal, &[]).expect("wave"); + assert!(first.events.is_empty(), "no release of a seat never parked"); + // Silent twice with nobody parked: the stall stands. + wave(&mut conductor, &journal, &[]).expect("nudged"); + let stalled = wave(&mut conductor, &journal, &[]); + assert!(matches!(stalled, Err(Error::Stalled { seats }) if seats == ["one"])); +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/support.rs b/crates/tinyhivemind-driver/src/conduct/test/support.rs index 6d8b8f3a..42cefbbc 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/support.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/support.rs @@ -211,6 +211,17 @@ pub(super) fn wave( conductor: &mut Conductor<'_, Seat>, journal: &Journal, calls: &[(&str, Vec)], +) -> Result { + wave_parking(conductor, journal, calls, &[]) +} + +/// A wave in which the seats in `parked` stop on the host instead of +/// recording anything. +pub(super) fn wave_parking( + conductor: &mut Conductor<'_, Seat>, + journal: &Journal, + calls: &[(&str, Vec)], + parked: &[&str], ) -> Result { let mut seen = Wave::default(); for step in conductor.begin_wave() { @@ -222,6 +233,10 @@ pub(super) fn wave( journal.thread(root) }); assert_eq!(brief.seat, turn.seat); + if parked.contains(&turn.seat.as_str()) { + conductor.record_parked(turn); + continue; + } let script = calls .iter() .find(|(seat, _)| *seat == turn.seat) diff --git a/crates/tinyhivemind-driver/src/conduct/test/wire.rs b/crates/tinyhivemind-driver/src/conduct/test/wire.rs index c0ed6a66..25fb0d33 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/wire.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/wire.rs @@ -180,6 +180,14 @@ fn every_event_and_refusal_survives_the_wire() { seat: "one".into(), thread: Some(Sequence(4)), }, + Event::Parked { + seat: "one".into(), + thread: None, + }, + Event::Resumed { + seat: "one".into(), + thread: Some(Sequence(4)), + }, Event::Broadcast { seat: "one".into(), to: vec!["two".into()], diff --git a/crates/tinyhivemind-driver/src/conduct/wave.rs b/crates/tinyhivemind-driver/src/conduct/wave.rs index 6293e0cf..c680f121 100644 --- a/crates/tinyhivemind-driver/src/conduct/wave.rs +++ b/crates/tinyhivemind-driver/src/conduct/wave.rs @@ -64,7 +64,7 @@ impl Wave { self.force_conclusions = nothing_due; } - fn event(&mut self, event: Event) { + pub(super) fn event(&mut self, event: Event) { self.steps.push_back(Step::Event(event)); } diff --git a/crates/tinyhivemind-driver/src/driver/brief.rs b/crates/tinyhivemind-driver/src/driver/brief.rs index 64f96c70..83700614 100644 --- a/crates/tinyhivemind-driver/src/driver/brief.rs +++ b/crates/tinyhivemind-driver/src/driver/brief.rs @@ -74,6 +74,24 @@ pub struct EpisodeBrief { pub awaiting: Vec, /// Handoffs held for this seat, delivered when it completes. pub queued: usize, + /// What the seat's other conversations hold, for a host that has them: + /// the newest rows of each, read as the seat, through the wave's + /// watermark. Context, not work: nothing in it is addressed here. + pub elsewhere: Vec, +} + +/// The newest rows of one conversation the seat is in that is not this +/// turn's, rendered by the host. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ElsewhereView { + /// The conversation's chat id. + pub chat: String, + /// Its display name. + pub name: String, + /// Its thread root, or `None` for that desk's open channel. + pub thread_root: Option, + /// Its newest rows, oldest first. + pub rows: Vec, } impl EpisodeBrief { @@ -103,6 +121,7 @@ impl EpisodeBrief { channel, new_rows, conversations, + elsewhere: Vec::new(), } } @@ -131,6 +150,7 @@ impl EpisodeBrief { fn render_desk(&self) -> String { let mut out = format!("## New desk messages\n{}", rows_or_nothing(&self.new_rows)); + out.push_str(&self.render_elsewhere()); let concluded: Vec = self .conversations .iter() @@ -207,18 +227,49 @@ impl EpisodeBrief { seat that asked you will ask them." }; format!( - "## A private conversation with @{other} (thread {})\n{}\n\n{role} Only the two of \ + "## A private conversation with @{other} (thread {})\n{}{}\n\n{role} Only the two of \ you read this thread.\n\nEvery tool call must carry \"chat\": \"{}\" and \ \"parent\": \"{}\". `ask` is not available inside a conversation. A `broadcast` made here \ hands work off on the desk, exactly as it would there.", root.0, rows_or_nothing(&self.new_rows), + self.render_elsewhere(), self.chat, root.0 ) } } +impl EpisodeBrief { + /// The seat's other conversations as a section, or nothing when the + /// host gave none. + fn render_elsewhere(&self) -> String { + if self.elsewhere.is_empty() { + return String::new(); + } + let mut out = String::from( + "\n\n## Elsewhere, for context\nWhat your other conversations hold. Nothing here is \ + addressed to you on this desk.", + ); + for view in &self.elsewhere { + let thread = view + .thread_root + .map(|root| format!(", thread {}", root.0)) + .unwrap_or_default(); + let _ = std::fmt::Write::write_fmt( + &mut out, + format_args!( + "\n\n### {} ({}{thread})\n{}", + view.name, + view.chat, + rows_or_nothing(&view.rows) + ), + ); + } + out + } +} + fn rows_or_nothing(rows: &[String]) -> String { if rows.is_empty() { "(nothing new)".to_owned() diff --git a/crates/tinyhivemind-driver/src/driver/brief/test.rs b/crates/tinyhivemind-driver/src/driver/brief/test.rs index 1e1e51da..fe5d8749 100644 --- a/crates/tinyhivemind-driver/src/driver/brief/test.rs +++ b/crates/tinyhivemind-driver/src/driver/brief/test.rs @@ -5,7 +5,7 @@ use tinyhivemind::Sequence; use tinyhivemind::speech::{Utterance, tool_specs}; -use super::{Channel, ConversationView, EpisodeBrief, standing_contract}; +use super::{Channel, ConversationView, ElsewhereView, EpisodeBrief, standing_contract}; use crate::driver::test::{committed, episode, hive}; use crate::driver::{CompletionDriver, DriverState}; @@ -191,3 +191,61 @@ fn the_standing_contract_is_the_specs_own_words_with_the_hosts_one_sentence() { ); assert!(text.contains("arguments {\"to\": ..., \"message\": ...}")); } + +#[test] +fn elsewhere_is_rendered_as_context_on_the_desk_and_in_a_thread_and_absent_when_empty() { + let (_hive, state) = state_with_an_ask(); + let mut desk = EpisodeBrief::for_turn( + &state, + "engineering", + "one", + Channel::Desk, + vec!["@operator: the task".into()], + Vec::new(), + ); + assert!(desk.elsewhere.is_empty(), "the episode fills nothing in"); + assert!(!desk.render().contains("Elsewhere")); + desk.elsewhere = vec![ + ElsewhereView { + chat: "marketing".into(), + name: "Marketing".into(), + thread_root: None, + rows: vec!["@three: launch is friday".into()], + }, + ElsewhereView { + chat: "marketing".into(), + name: "Marketing".into(), + thread_root: Some(Sequence(9)), + rows: Vec::new(), + }, + ]; + let text = desk.render(); + let rows_at = text.find("@operator: the task").expect("new rows"); + let elsewhere_at = text.find("## Elsewhere, for context").expect("section"); + assert!(rows_at < elsewhere_at, "the turn's own rows come first"); + assert!(text.contains("### Marketing (marketing)\n@three: launch is friday")); + assert!(text.contains("### Marketing (marketing, thread 9)\n(nothing new)")); + assert!(text.contains("Nothing here is addressed to you on this desk")); + + let mut thread = EpisodeBrief::for_turn( + &state, + "engineering", + "two", + Channel::Thread { + root: Sequence(1), + other: "one".into(), + opened_it: false, + }, + vec!["@one: ?".into()], + Vec::new(), + ); + thread.elsewhere = vec![ElsewhereView { + chat: "legal".into(), + name: "Legal".into(), + thread_root: None, + rows: vec!["@four: cleared".into()], + }]; + let text = thread.render(); + assert!(text.find("@one: ?").expect("rows") < text.find("### Legal (legal)").expect("section")); + assert!(text.contains("A peer asked you this")); +} diff --git a/crates/tinyhivemind-driver/src/driver/mod.rs b/crates/tinyhivemind-driver/src/driver/mod.rs index 875b31b0..d08ed2d9 100644 --- a/crates/tinyhivemind-driver/src/driver/mod.rs +++ b/crates/tinyhivemind-driver/src/driver/mod.rs @@ -19,7 +19,7 @@ use tinyhivemind_hive::{ }; use crate::{BoundHive, Error, Result}; -pub use brief::{Channel, ConversationView, EpisodeBrief, standing_contract}; +pub use brief::{Channel, ConversationView, ElsewhereView, EpisodeBrief, standing_contract}; #[cfg(test)] use broadcast::route_ids; pub use ledger::{AssignmentSpend, Handoff, Ledger, Seen}; diff --git a/crates/tinyhivemind-driver/src/error/mod.rs b/crates/tinyhivemind-driver/src/error/mod.rs index 33c644db..a889712e 100644 --- a/crates/tinyhivemind-driver/src/error/mod.rs +++ b/crates/tinyhivemind-driver/src/error/mod.rs @@ -248,6 +248,13 @@ pub enum Error { /// The seats holding it. seats: Vec, }, + /// Nothing is due anywhere and every seat that could move is parked on + /// the host, which released none of them. + #[error("episode parked on {seats:?} and the host released none")] + Parked { + /// The seats parked. + seats: Vec, + }, /// The episode ran past its turn wall. #[error("turn wall of {wall} reached")] TurnWall { diff --git a/crates/tinyhivemind-driver/src/lib.rs b/crates/tinyhivemind-driver/src/lib.rs index ebbe9ffb..1ae94c6b 100644 --- a/crates/tinyhivemind-driver/src/lib.rs +++ b/crates/tinyhivemind-driver/src/lib.rs @@ -82,8 +82,8 @@ pub use conduct::{ }; pub use driver::{ AssignmentSpend, BroadcastRouting, Channel, CommittedUtterance, CompletionDriver, - ConversationView, DriverState, EpisodeBrief, Handoff, HostAction, Ledger, PendingAgent, - PendingRound, Seen, Transition, standing_contract, + ConversationView, DriverState, ElsewhereView, EpisodeBrief, Handoff, HostAction, Ledger, + PendingAgent, PendingRound, Seen, Transition, standing_contract, }; pub use error::{Error, Result}; pub use graph::{AgentBinding, BoundAgent, BoundHive, HiveGraph}; From a34f6b308574c95c82eba8e3647f5e7e6079bd73 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 23 Sep 2026 02:29:09 +0530 Subject: [PATCH 2/3] Read what a seat's other conversations hold A host that briefs a seat on its other channels writes that read itself, and a read written in the host is how a seat ends up shown a row it was never addressed on. `gather_elsewhere` does it over the same log port and the same projection as every other read: narrowed to the seat, every conversation bounded by one `before` so a turn's context is a snapshot, and a conversation the seat may read nothing of listed with no rows rather than dropped. Co-Authored-By: Claude Opus 5 --- crates/tinyhivemind/src/README.md | 1 + crates/tinyhivemind/src/elsewhere/README.md | 43 ++++ crates/tinyhivemind/src/elsewhere/mod.rs | 84 +++++++ crates/tinyhivemind/src/elsewhere/test.rs | 257 ++++++++++++++++++++ crates/tinyhivemind/src/elsewhere/types.rs | 30 +++ crates/tinyhivemind/src/lib.rs | 2 + 6 files changed, 417 insertions(+) create mode 100644 crates/tinyhivemind/src/elsewhere/README.md create mode 100644 crates/tinyhivemind/src/elsewhere/mod.rs create mode 100644 crates/tinyhivemind/src/elsewhere/test.rs create mode 100644 crates/tinyhivemind/src/elsewhere/types.rs diff --git a/crates/tinyhivemind/src/README.md b/crates/tinyhivemind/src/README.md index cfba42a5..9c99a78a 100644 --- a/crates/tinyhivemind/src/README.md +++ b/crates/tinyhivemind/src/README.md @@ -9,6 +9,7 @@ about a live session; see its own `README.md` for the how and why. | [`session`](session) | How does a turn walk a host-owned, globally sequenced log into an attributed, audience-filtered transcript? | | [`briefing`](briefing) | What ephemeral context (teammates, coordination rules, history, threads, pins) does one viewer's turn open with? | | [`approval`](approval) | How does a pure ask decision reach one host-owned atomic human approval boundary? | +| [`elsewhere`](elsewhere) | What do this seat's *other* conversations hold, for the turn it is taking in this one? | | [`sharing`](sharing) | How does a host hand an already-briefed session only what changed since its last watermark, instead of re-briefing it? | | [`search`](search) | How does a turn reach a message or thread outside its window, on request? | | [`pins`](pins) | Which messages does every turn see whether or not it asked? | diff --git a/crates/tinyhivemind/src/elsewhere/README.md b/crates/tinyhivemind/src/elsewhere/README.md new file mode 100644 index 00000000..6d447d97 --- /dev/null +++ b/crates/tinyhivemind/src/elsewhere/README.md @@ -0,0 +1,43 @@ +# `elsewhere` + +What a seat's *other* conversations hold, for the turn it is taking in this +one. + +A turn is shown its own channel by whoever runs it. This module answers the +rest: for a seat about to speak in one conversation, the newest rows of +every other conversation it is in, read as that seat. It is the read a host +would otherwise write itself, and writing it in the host is how a seat ends +up reading rows it was never addressed on — so it lives here, over the same +`SessionLog` port and the same projection as every other read. + +`gather_elsewhere` takes an `ElsewhereQuery`: the seat, every conversation +it is in, the one its turn is in (skipped, and `None` skips nothing), an +exclusive `before` bound, and a window. It returns one `Elsewhere` per +conversation read, each holding that conversation's projected rows. + +Three properties are the point: + +- **Narrowed to the seat.** Every read projects as `Viewer::Agent`, so a + private row reaches the seat elsewhere exactly when it would reach it + there — never more. +- **One moment.** `before` bounds every conversation's read, so a turn's + context is a snapshot rather than a set of reads drifting row by row + while the log grows. +- **Nothing is dropped.** A conversation the seat may read nothing of comes + back with no rows rather than being left out, so a caller that listed its + channels gets the same list back and can say "nothing new" about one. + +This module stores nothing and decides nothing. What the rows mean for a +turn is the caller's: `tinyhivemind-driver` carries them in +`EpisodeBrief::elsewhere` and renders them under a heading that says they +are context, not work. + +`render_row` is the one-line rendering every caller uses to put a row in +front of a model — `@author: content`, and nothing for a row the viewer may +not read. + +| file | holds | +| --- | --- | +| `mod.rs` | `gather_elsewhere`, `render_row` | +| `types.rs` | `ElsewhereQuery`, `Elsewhere` | +| `test.rs` | the skip, the narrowing, the bound, a thread as its own conversation, and a failed read | diff --git a/crates/tinyhivemind/src/elsewhere/mod.rs b/crates/tinyhivemind/src/elsewhere/mod.rs new file mode 100644 index 00000000..06f0b657 --- /dev/null +++ b/crates/tinyhivemind/src/elsewhere/mod.rs @@ -0,0 +1,84 @@ +//! What a seat's *other* conversations hold, for the turn it is taking in +//! this one. + +#[cfg(test)] +mod test; + +mod types; + +pub use types::{Elsewhere, ElsewhereQuery}; + +use crate::aside::Viewer; +use crate::{ + Conversation, Result, SessionAuthor, SessionLog, SessionMessage, SessionQuery, project_session, +}; + +/// Read the newest rows of every conversation in `query.conversations` +/// except the one the turn is in, as `seat` reads them. +/// +/// A turn is shown its own channel by whoever runs it. This is the rest of +/// what the seat would know if it were reading: one page per conversation, +/// projected as the seat, so a row it was not addressed on is withheld +/// exactly as it is everywhere else. Nothing here is work -- it is context, +/// and a caller that renders it says so. +/// +/// `before` bounds every read, so a set of conversations read for one turn +/// is read as of one moment rather than drifting row by row. A conversation +/// the seat may read nothing of is returned with no rows rather than +/// dropped: a caller that lists its channels gets the same list back. +/// +/// # Errors +/// +/// Returns [`Error::Read`](crate::Error::Read) for a host read failure, or a +/// page-validation error when a host breaks the port's contract. +pub async fn gather_elsewhere( + log: &dyn SessionLog, + query: &ElsewhereQuery<'_>, +) -> Result> { + let mut gathered = Vec::new(); + for conversation in query.conversations { + if same_conversation(conversation, query.current) { + continue; + } + let rows = project_session( + log, + &SessionQuery { + conversation: conversation.clone(), + viewer: Viewer::Agent { + id: query.seat.to_owned(), + }, + before: query.before, + window: query.window, + }, + ) + .await?; + gathered.push(Elsewhere { + conversation: conversation.clone(), + rows, + }); + } + Ok(gathered) +} + +/// Whether two conversations are the same desk and the same thread. +fn same_conversation(one: &Conversation, other: Option<&Conversation>) -> bool { + other.is_some_and(|other| one.desk_id == other.desk_id && one.thread_root == other.thread_root) +} + +/// One row as a model reads it: `@author: content`, or nothing for a row +/// withheld from the viewer the projection ran as. +/// +/// Every caller that puts rows in front of a model renders them this way, +/// so the rows a seat reads from elsewhere look like the rows it reads +/// here. +#[must_use] +pub fn render_row(row: &SessionMessage) -> Option { + let content = row.readable()?; + let author = match &row.author { + SessionAuthor::Operator => "operator", + SessionAuthor::Agent { label, .. } + | SessionAuthor::Person { label, .. } + | SessionAuthor::System { label, .. } => label, + }; + Some(format!("@{author}: {content}")) +} diff --git a/crates/tinyhivemind/src/elsewhere/test.rs b/crates/tinyhivemind/src/elsewhere/test.rs new file mode 100644 index 00000000..ce4f63ef --- /dev/null +++ b/crates/tinyhivemind/src/elsewhere/test.rs @@ -0,0 +1,257 @@ +//! What a seat reads of its other conversations: every one but the turn's, +//! narrowed to the seat, bounded by one moment. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use std::sync::Mutex; + +use super::{Elsewhere, ElsewhereQuery, gather_elsewhere, render_row}; +use crate::aside::Audience; +use crate::{ + Conversation, Error, LogMessage, Sequence, SessionAuthor, SessionFuture, SessionLog, + SessionPage, +}; + +/// A journal of rows across several desks, read newest-first. +struct Rows(Vec, Mutex); + +impl Rows { + fn new(rows: Vec) -> Self { + Self(rows, Mutex::new(0)) + } + + fn reads(&self) -> usize { + *self.1.lock().expect("reads") + } +} + +impl SessionLog for Rows { + fn read_before(&self, before: Option, limit: usize) -> SessionFuture<'_> { + *self.1.lock().expect("reads") += 1; + let mut older: Vec = self + .0 + .iter() + .filter(|row| before.is_none_or(|bound| row.sequence < bound)) + .cloned() + .collect(); + older.sort_by_key(|row| std::cmp::Reverse(row.sequence)); + let taken: Vec = older.iter().take(limit).cloned().collect(); + let next_before = (older.len() > taken.len()) + .then(|| taken.last().map(|row| row.sequence)) + .flatten(); + Box::pin(async move { + Ok(SessionPage { + messages: taken, + next_before, + }) + }) + } +} + +/// A log whose only read fails. +struct Broken; + +impl SessionLog for Broken { + fn read_before(&self, _before: Option, _limit: usize) -> SessionFuture<'_> { + Box::pin(async { Err(Box::new(std::io::Error::other("offline")) as crate::SourceError) }) + } +} + +fn agent(id: &str) -> SessionAuthor { + SessionAuthor::Agent { + id: id.to_owned(), + label: id.to_owned(), + } +} + +fn row(sequence: u64, desk: &str, author: &str, content: &str, audience: Audience) -> LogMessage { + LogMessage { + sequence: Sequence(sequence), + chat_id: Some(desk.to_owned()), + parent: None, + author: agent(author), + content: content.to_owned(), + audience, + } +} + +fn desk(id: &str) -> Conversation { + Conversation { + desk_id: id.to_owned(), + desk_name: id.to_owned(), + thread_root: None, + } +} + +fn run(future: F) -> F::Output { + tokio::runtime::Builder::new_current_thread() + .build() + .expect("runtime") + .block_on(future) +} + +fn log() -> Rows { + Rows::new(vec![ + row(1, "engineering", "one", "the deploy is out", Audience::Desk), + row(2, "marketing", "three", "launch is friday", Audience::Desk), + row( + 3, + "marketing", + "three", + "budget is between us", + Audience::Aside { + members: vec!["four".to_owned()], + }, + ), + row(4, "legal", "five", "cleared", Audience::Desk), + row(5, "marketing", "four", "noted", Audience::Desk), + ]) +} + +#[test] +fn every_conversation_but_the_turn_is_read_as_the_seat() { + let log = log(); + let channels = [desk("engineering"), desk("marketing"), desk("legal")]; + let here = desk("engineering"); + let gathered = run(gather_elsewhere( + &log, + &ElsewhereQuery { + seat: "one", + conversations: &channels, + current: Some(&here), + before: None, + window: 30, + }, + )) + .expect("reads"); + let desks: Vec<&str> = gathered + .iter() + .map(|found| found.conversation.desk_id.as_str()) + .collect(); + assert_eq!( + desks, + ["marketing", "legal"], + "the turn's own desk is skipped" + ); + let rendered: Vec = gathered[0].rows.iter().filter_map(render_row).collect(); + assert_eq!( + rendered, + ["@three: launch is friday", "@four: noted"], + "the aside to four is withheld from one" + ); + // Read as four, the aside is there. + let theirs = run(gather_elsewhere( + &log, + &ElsewhereQuery { + seat: "four", + conversations: &channels, + current: Some(&here), + before: None, + window: 30, + }, + )) + .expect("reads"); + let rendered: Vec = theirs[0].rows.iter().filter_map(render_row).collect(); + assert!( + rendered.contains(&"@three: budget is between us".to_owned()), + "{rendered:?}" + ); +} + +#[test] +fn a_bound_holds_every_conversation_to_one_moment_and_no_current_reads_them_all() { + let log = log(); + let channels = [desk("marketing"), desk("legal")]; + let gathered = run(gather_elsewhere( + &log, + &ElsewhereQuery { + seat: "one", + conversations: &channels, + current: None, + before: Some(Sequence(4)), + window: 30, + }, + )) + .expect("reads"); + assert_eq!(gathered.len(), 2, "nothing is skipped without a current"); + let rendered: Vec = gathered[0].rows.iter().filter_map(render_row).collect(); + assert_eq!( + rendered, + ["@three: launch is friday"], + "row 5 is above the bound" + ); + assert!( + gathered[1].rows.is_empty(), + "legal's only row is at the bound, which is exclusive" + ); + assert_eq!( + gathered[1], + Elsewhere { + conversation: desk("legal"), + rows: Vec::new(), + }, + "a conversation with nothing to show is still listed" + ); +} + +#[test] +fn a_thread_is_its_own_conversation_and_a_failed_read_is_reported() { + let log = Rows::new(vec![ + LogMessage { + sequence: Sequence(1), + chat_id: Some("engineering".into()), + parent: None, + author: agent("one"), + content: "which port?".into(), + audience: Audience::Aside { + members: vec!["two".to_owned()], + }, + }, + LogMessage { + sequence: Sequence(2), + chat_id: Some("engineering".into()), + parent: Some(Sequence(1)), + author: agent("two"), + content: "8080".into(), + audience: Audience::Aside { + members: vec!["two".to_owned()], + }, + }, + ]); + let thread = Conversation { + thread_root: Some(Sequence(1)), + ..desk("engineering") + }; + let channels = [desk("engineering"), thread.clone()]; + let gathered = run(gather_elsewhere( + &log, + &ElsewhereQuery { + seat: "two", + conversations: &channels, + current: Some(&thread), + before: None, + window: 30, + }, + )) + .expect("reads"); + assert_eq!( + gathered.len(), + 1, + "the thread it is in is skipped, the desk is not" + ); + assert_eq!(gathered[0].conversation.thread_root, None); + assert!(log.reads() >= 1); + assert!(format!("{gathered:?}").contains("engineering")); + + let broken = run(gather_elsewhere( + &Broken, + &ElsewhereQuery { + seat: "one", + conversations: &channels, + current: None, + before: None, + window: 30, + }, + )); + assert!(matches!(broken, Err(Error::Read { .. })), "{broken:?}"); +} diff --git a/crates/tinyhivemind/src/elsewhere/types.rs b/crates/tinyhivemind/src/elsewhere/types.rs new file mode 100644 index 00000000..51c771d7 --- /dev/null +++ b/crates/tinyhivemind/src/elsewhere/types.rs @@ -0,0 +1,30 @@ +//! What to read, and what came back. + +use crate::{Conversation, SessionMessage}; + +/// Which conversations to read for a seat, as of when. +#[derive(Clone, Debug)] +pub struct ElsewhereQuery<'a> { + /// The seat the rows are read as. + pub seat: &'a str, + /// Every conversation the seat is in, including the one it is taking a + /// turn in: that one is skipped rather than having to be left out. + pub conversations: &'a [Conversation], + /// The conversation the turn is in, skipped. `None` reads them all, + /// which is what a caller briefing a seat outside a turn wants. + pub current: Option<&'a Conversation>, + /// Exclusive upper bound on every read, so one turn's context is read + /// as of one moment. + pub before: Option, + /// Rows per conversation. + pub window: usize, +} + +/// One conversation's newest rows, as the seat reads them. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct Elsewhere { + /// Which conversation. + pub conversation: Conversation, + /// Its rows, chronological, narrowed to what the seat may read. + pub rows: Vec, +} diff --git a/crates/tinyhivemind/src/lib.rs b/crates/tinyhivemind/src/lib.rs index 7197ee9b..ca03a851 100644 --- a/crates/tinyhivemind/src/lib.rs +++ b/crates/tinyhivemind/src/lib.rs @@ -57,6 +57,7 @@ pub mod approval; pub mod briefing; pub mod digest; pub mod dispatch; +pub mod elsewhere; pub mod error; pub mod pins; pub mod referral; @@ -87,6 +88,7 @@ pub use dispatch::{ EnqueueOutcome, EnqueueRefusal, MentionDispatchOutcome, MentionTurnFuture, MentionTurnQueue, dispatch_mention, }; +pub use elsewhere::{Elsewhere, ElsewhereQuery, gather_elsewhere, render_row}; pub use error::{Error, Result}; pub use pins::{ PIN_EXCERPT_CHARS, PIN_LIMIT, PIN_SCAN, Pin, PinAction, PinDirective, fold_pins, pin_note, From ab7234137d8cc2539ba6032215a67438c1a69d2e Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 23 Sep 2026 02:37:44 +0530 Subject: [PATCH 3/3] Carry parking and cross-channel context through the episode loop `TurnResult` becomes an enum -- replied, failed, parked -- which drops a `None` case no runner ever produced and gives a turn that stopped on the host somewhere to say so. `EpisodeHost::after_turn` returns a `Disposition`, so a host that queues an approval parks the seat; the loop records what the turn called and holds it. With nothing to run and seats parked, the loop asks `Journal::released`, where a host blocks on its own queue, and a host that releases nobody ends the episode parked rather than spinning. `Journal::channels` names the seat's other conversations, whose newest rows are gathered through the same wave watermark and carried in the brief as context. Co-Authored-By: Claude Opus 5 --- crates/tinyhivemind-driver/src/conduct/mod.rs | 15 +- .../src/conduct/test/parked.rs | 43 +++- .../src/conduct/test/support.rs | 12 +- .../tinyhivemind-openhuman/src/embed/mod.rs | 10 +- .../src/episode/README.md | 21 +- .../tinyhivemind-openhuman/src/episode/mod.rs | 208 ++++++++++++++---- .../src/episode/test/flow.rs | 13 +- .../src/episode/test/mod.rs | 1 + .../src/episode/test/parking.rs | 158 +++++++++++++ .../src/episode/test/support.rs | 42 +++- .../src/hosted/README.md | 6 +- .../tinyhivemind-openhuman/src/hosted/mod.rs | 38 +++- .../src/journal/README.md | 4 +- .../tinyhivemind-openhuman/src/journal/mod.rs | 34 ++- crates/tinyhivemind-openhuman/src/lib.rs | 4 +- crates/tinyhivemind-openhuman/src/raw/mod.rs | 10 +- .../src/runner/README.md | 3 +- .../tinyhivemind-openhuman/src/runner/mod.rs | 35 ++- .../tinyhivemind-openhuman/src/runner/test.rs | 26 ++- .../openhuman/src/bin/conducted/hosted.rs | 23 +- examples/openhuman/src/bin/deepswe_hive.rs | 6 +- examples/openhuman/src/bin/pe1006_hive.rs | 8 +- examples/openhuman/src/main.rs | 46 ++-- 23 files changed, 618 insertions(+), 148 deletions(-) create mode 100644 crates/tinyhivemind-openhuman/src/episode/test/parking.rs diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs index 0721c510..c3799e7b 100644 --- a/crates/tinyhivemind-driver/src/conduct/mod.rs +++ b/crates/tinyhivemind-driver/src/conduct/mod.rs @@ -367,15 +367,16 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { } /// A turn stopped on something only the host can settle -- an approval, - /// typically -- and recorded nothing. The seat is held where it parked: - /// it is not nudged for silence, does not stall the episode, and is not - /// proposed again until the host releases it. A parked askee's - /// conversation waits with it. - pub fn record_parked(&mut self, turn: &Turn) { - self.turns += 1; + /// typically. What it called before it stopped is recorded as any turn's + /// calls are; the seat is then held where it parked: not nudged for + /// silence, not counted toward a stall, and not proposed again until the + /// host releases it. A parked askee's conversation waits with it. + pub fn record_parked(&mut self, turn: &Turn, calls: impl IntoIterator) { + self.record(turn, calls); let thread = turn.thread(); if let Some(child) = thread.and_then(|root| self.children.get_mut(&root)) { - // Not a silence: the askee did not take this turn, for nudging. + // Not a silence: the askee is coming back to this conversation, + // so it is not nudged for having said nothing in it. child.turned = false; } self.parked.insert(turn.seat.clone(), thread); diff --git a/crates/tinyhivemind-driver/src/conduct/test/parked.rs b/crates/tinyhivemind-driver/src/conduct/test/parked.rs index efaf3e53..8e891f21 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/parked.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/parked.rs @@ -3,7 +3,9 @@ #![allow(clippy::expect_used, clippy::unwrap_used)] -use super::support::{Journal, ask, complete, hive, policy, seats, two_seat, wave, wave_parking}; +use super::support::{ + Journal, ask, broadcast, complete, hive, policy, seats, two_seat, wave, wave_parking, +}; use crate::conduct::{ConductPolicy, Conductor, Event}; use crate::driver::BroadcastRouting; use crate::{CompletionDriver, Error}; @@ -159,3 +161,42 @@ fn releasing_a_seat_that_is_not_parked_changes_nothing_and_a_stall_is_still_a_st let stalled = wave(&mut conductor, &journal, &[]); assert!(matches!(stalled, Err(Error::Stalled { seats }) if seats == ["one"])); } + +#[test] +fn what_a_seat_said_before_it_parked_is_recorded() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + // One hands work off and then stops on the host: the broadcast lands. + let parked = wave_parking( + &mut conductor, + &journal, + &[("one", vec![broadcast("someone take the migration")])], + &["one"], + ) + .expect("wave"); + assert!( + journal + .bodies() + .iter() + .any(|body| body.contains("take the migration")), + "{:?}", + journal.bodies() + ); + assert!( + parked + .events + .iter() + .any(|event| matches!(event, Event::Parked { seat, .. } if seat == "one")) + ); + assert_eq!(conductor.parked(), vec!["one".to_owned()]); +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/support.rs b/crates/tinyhivemind-driver/src/conduct/test/support.rs index 42cefbbc..bce231c6 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/support.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/support.rs @@ -223,6 +223,7 @@ pub(super) fn wave_parking( calls: &[(&str, Vec)], parked: &[&str], ) -> Result { + // A parked seat's calls, if it made any, are recorded before it is held. let mut seen = Wave::default(); for step in conductor.begin_wave() { take(step, journal, &mut seen); @@ -233,16 +234,17 @@ pub(super) fn wave_parking( journal.thread(root) }); assert_eq!(brief.seat, turn.seat); - if parked.contains(&turn.seat.as_str()) { - conductor.record_parked(turn); - continue; - } let script = calls .iter() .find(|(seat, _)| *seat == turn.seat) .map(|(_, calls)| calls.clone()) .unwrap_or_default(); - conductor.record(turn, script.into_iter().map(ToolCall::Speak)); + let said = script.into_iter().map(ToolCall::Speak); + if parked.contains(&turn.seat.as_str()) { + conductor.record_parked(turn, said); + } else { + conductor.record(turn, said); + } } seen.turns = turns; while let Some(step) = conductor.step()? { diff --git a/crates/tinyhivemind-openhuman/src/embed/mod.rs b/crates/tinyhivemind-openhuman/src/embed/mod.rs index d2d7bcbc..024d94d7 100644 --- a/crates/tinyhivemind-openhuman/src/embed/mod.rs +++ b/crates/tinyhivemind-openhuman/src/embed/mod.rs @@ -27,8 +27,8 @@ use openhuman_embed::{ use tinyhivemind_driver::{AgentBinding, BoundAgent}; use tinyhivemind_mcp::{EpisodeTools, Server, serve}; -use crate::Result; -use crate::runner::{Lane, SeatRunner, TURN_TIMEOUT, TurnJob, unseated}; +use crate::runner::{Lane, SeatRunner, TURN_TIMEOUT, TurnJob, TurnResult, unseated}; +use crate::{Error, Result}; use tinyhivemind::Sequence; /// An `openhuman-embed` agent as the handle the driver binds. @@ -132,9 +132,9 @@ impl SeatRunner for EmbedRunner { ) .await { - Ok(Ok(outcome)) => Some(Ok(outcome.reply)), - Ok(Err(error)) => Some(Err(error.to_string())), - Err(_) => None, + Ok(Ok(outcome)) => TurnResult::Replied(outcome.reply), + Ok(Err(error)) => TurnResult::Failed(error.to_string()), + Err(_) => TurnResult::Failed(Error::TimedOut { seat: seat.clone() }.to_string()), }; (seat, lane, result) }) diff --git a/crates/tinyhivemind-openhuman/src/episode/README.md b/crates/tinyhivemind-openhuman/src/episode/README.md index efcf7e44..c69c9f6f 100644 --- a/crates/tinyhivemind-openhuman/src/episode/README.md +++ b/crates/tinyhivemind-openhuman/src/episode/README.md @@ -9,9 +9,22 @@ 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` +and five 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. +reply, refusals and recorded calls, `channels` to name the seat's other +conversations, and `released` to say which parked seats the host has +settled. `Report` is what an episode came to. + +A turn that comes back `TurnResult::Parked` is recorded with whatever it +called and then held: the conductor stops proposing that seat. When a wave +has nothing to run and seats are parked, the loop asks `released`, which is +where a host blocks on its own approval queue; a host that releases nobody +ends the episode with `Error::Parked` rather than spinning. + +`channels` names every conversation the seat is in that this episode does +not run. Their newest rows are read through `gather_elsewhere`, bounded by +the same wave watermark as every other read, and carried in +`EpisodeBrief::elsewhere` under a heading that says they are context. 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 @@ -27,5 +40,5 @@ 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) | +| `mod.rs` | `Journal`, `Report`, `Released`, `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), `parking.rs` (a seat held on the host, and its other conversations in its brief), `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 index 8bfa50f9..24d0b31e 100644 --- a/crates/tinyhivemind-openhuman/src/episode/mod.rs +++ b/crates/tinyhivemind-openhuman/src/episode/mod.rs @@ -20,14 +20,16 @@ #[cfg(test)] mod test; +use std::pin::Pin; + use tinyhivemind::aside::Viewer; use tinyhivemind::{ - Conversation, SESSION_WINDOW, Sequence, SessionAuthor, SessionLog, SessionMessage, - SessionQuery, project_session, + Conversation, ElsewhereQuery, SESSION_WINDOW, Sequence, SessionAuthor, SessionLog, + SessionMessage, SessionQuery, gather_elsewhere, project_session, }; use tinyhivemind_driver::{ BoundAgent, BroadcastRouting, Commit, CompletionDriver, ConductPolicy, Conductor, Door, - EpisodeBrief, Event, Note, Step, + ElsewhereView, EpisodeBrief, Event, Note, Step, Turn, }; use tinyhivemind_tools::{Dispatch, Refusal}; @@ -65,6 +67,36 @@ pub trait Journal: Send + Sync { let _ = event; } + /// Every conversation `seat` is in that this episode does not run: its + /// other desks, and any thread of them. The newest rows of each are + /// read as the seat and carried in its brief as context. The default is + /// none, and an episode is then the only thing a seat is shown. + /// + /// This desk may be named here too: the turn's own conversation is + /// skipped, so a thread turn is shown the desk it hangs off and a desk + /// turn is not shown itself. + fn channels(&self, seat: &str) -> Vec { + let _ = seat; + Vec::new() + } + + /// Nothing is due and these seats are parked: the seats the host has + /// settled, waiting until it has one. + /// + /// The episode cannot go on until one comes back, so a host that parks + /// blocks here on its own queue -- an approval being answered -- and + /// returns the seats it released. Returning none ends the episode with + /// [`driver::Error::Parked`](tinyhivemind_driver::Error::Parked), which + /// is the default, because a host that never parks is never asked. + /// + /// # Errors + /// + /// Whatever stops the host waiting. + fn released<'a>(&'a self, parked: &'a [String]) -> Released<'a> { + let _ = parked; + Box::pin(async { Ok(Vec::new()) }) + } + /// 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 { @@ -86,6 +118,9 @@ pub trait Journal: Send + Sync { } } +/// The seats a host released, once it has any. +pub type Released<'a> = Pin>> + Send + 'a>>; + /// What one episode came to. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct Report { @@ -134,7 +169,10 @@ where for step in conductor.begin_wave() { settle(journal, &mut conductor, step).await?; } - let turns = conductor.turns()?; + let mut turns = conductor.turns()?; + if turns.is_empty() { + turns = wait_for_release(journal, &mut conductor).await?; + } // 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 @@ -142,44 +180,7 @@ where 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)); + jobs.push(open_turn(journal, runner, &mut conductor, &desk, turn, latest).await?); } let named: Vec<(String, Lane)> = turns .iter() @@ -197,7 +198,12 @@ where 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)); + let calls = events.into_iter().map(|event| event.call); + if outcome.parked() { + conductor.record_parked(turn, calls); + } else { + conductor.record(turn, calls); + } } } while let Some(step) = conductor.step()? { @@ -254,7 +260,7 @@ async fn join_turns( done.push(( seat, lane, - Some(Err(format!("the turn's task failed: {error}"))), + TurnResult::Failed(format!("the turn's task failed: {error}")), )); } } @@ -262,6 +268,120 @@ async fn join_turns( done } +/// Nothing is due: if seats are held on the host, wait for it to release +/// one and ask again; otherwise the wave is simply empty. +async fn wait_for_release( + journal: &J, + conductor: &mut Conductor<'_, A>, +) -> Result> { + let parked = conductor.parked(); + if parked.is_empty() { + return Ok(Vec::new()); + } + let released = journal.released(&parked).await?; + if released.is_empty() { + return Err(tinyhivemind_driver::Error::Parked { seats: parked }.into()); + } + for seat in &released { + conductor.resume_seat(seat); + } + Ok(conductor.turns()?) +} + +/// One turn opened: the rows it has not seen, the record, the brief, the +/// prompt, and the job started on the runner. +async fn open_turn( + journal: &J, + runner: &R, + conductor: &mut Conductor<'_, A>, + desk: &Conversation, + turn: &Turn, + latest: Option, +) -> Result { + let channel = Conversation { + thread_root: turn.thread(), + ..desk.clone() + }; + let log = journal.log(); + let rows = rows_above(log, &channel, &turn.seat, turn.since, latest).await?; + let window = match turn.thread() { + None => rows.clone(), + Some(_) => rows_above(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() + }; + transcripts.insert( + root, + rows_above(log, &thread, &turn.seat, None, latest).await?, + ); + } + let mut brief = conductor.open_turn(turn, latest, rows, |root| { + transcripts.get(&root).cloned().unwrap_or_default() + }); + brief.elsewhere = elsewhere(journal, &turn.seat, &channel, latest).await?; + let prompt = journal.compose(&turn.seat, &brief); + let lane = turn.thread().map_or(Lane::Desk, Lane::Thread); + Ok(runner.turn(turn.seat.clone(), lane, turn.since, prompt)) +} + +/// What the seat's other conversations hold, as the brief carries them: +/// every channel the host names but the turn's own, read as the seat, +/// through the wave's watermark. +async fn elsewhere( + journal: &J, + seat: &str, + current: &Conversation, + latest: Option, +) -> Result> { + let channels = journal.channels(seat); + if channels.is_empty() { + return Ok(Vec::new()); + } + let Some(latest) = latest else { + return Ok(Vec::new()); + }; + let gathered = gather_elsewhere( + journal.log(), + &ElsewhereQuery { + seat, + conversations: &channels, + current: Some(current), + // Exclusive, so one above the watermark, as every other read + // of this wave is bounded. + before: latest.0.checked_add(1).map(Sequence), + window: SESSION_WINDOW, + }, + ) + .await?; + Ok(gathered + .into_iter() + .map(|found| ElsewhereView { + chat: found.conversation.desk_id, + name: found.conversation.desk_name, + thread_root: found.conversation.thread_root, + rows: found.rows.iter().filter_map(render).collect(), + }) + .collect()) +} + /// The newest sequence in the log, or `None` for a log with no rows. async fn latest(log: &dyn SessionLog) -> Result> { let page = log diff --git a/crates/tinyhivemind-openhuman/src/episode/test/flow.rs b/crates/tinyhivemind-openhuman/src/episode/test/flow.rs index 2d883397..522d69ac 100644 --- a/crates/tinyhivemind-openhuman/src/episode/test/flow.rs +++ b/crates/tinyhivemind-openhuman/src/episode/test/flow.rs @@ -10,7 +10,7 @@ use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Eve 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; +use crate::runner::{Lane, TurnResult}; #[test] fn an_episode_runs_from_its_door_to_quiescence_over_the_journal() { @@ -120,7 +120,7 @@ fn the_journal_saw_each_turn(journal: &TestJournal, runner: &ScriptRunner) { assert!( turns .iter() - .all(|(_, _, outcome, _, _)| matches!(outcome, Some(Ok(_)))) + .all(|(_, _, outcome, _, _)| matches!(outcome, TurnResult::Replied(_))) ); // `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. @@ -176,11 +176,10 @@ fn a_seat_that_says_nothing_is_nudged_and_then_the_episode_stalls() { .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!(turns.iter().any(|(_, _, outcome, _, recorded)| matches!( + outcome, + TurnResult::Failed(_) + ) && *recorded == 0)); assert!( journal .events() diff --git a/crates/tinyhivemind-openhuman/src/episode/test/mod.rs b/crates/tinyhivemind-openhuman/src/episode/test/mod.rs index 30b2af4b..fe85d310 100644 --- a/crates/tinyhivemind-openhuman/src/episode/test/mod.rs +++ b/crates/tinyhivemind-openhuman/src/episode/test/mod.rs @@ -2,5 +2,6 @@ mod flow; mod journals; +mod parking; mod support; mod watermark; diff --git a/crates/tinyhivemind-openhuman/src/episode/test/parking.rs b/crates/tinyhivemind-openhuman/src/episode/test/parking.rs new file mode 100644 index 00000000..81847fbd --- /dev/null +++ b/crates/tinyhivemind-openhuman/src/episode/test/parking.rs @@ -0,0 +1,158 @@ +//! A seat parked on the host: held, waited for, released; and what its +//! other conversations put in its brief. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use std::collections::VecDeque; + +use tinyhivemind::{Conversation, Sequence}; +use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Event}; + +use super::super::run_episode; +use super::support::{ScriptRunner, TestJournal, complete, door, hive, policy, run}; +use crate::Error; + +/// A call that makes the scripted turn park rather than reply. +fn park() -> (&'static str, serde_json::Value) { + ("park", serde_json::Value::Null) +} + +#[test] +fn a_parked_seat_waits_for_the_host_and_runs_again_once_released() { + 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(); + *journal.release.lock().unwrap() = VecDeque::from([vec!["one".to_owned()]]); + // One parks on its first turn, then completes on the turn it is given + // after the host releases it. + let runner = ScriptRunner::new( + &["one", "two"], + &[("one", vec![vec![park()], vec![complete("approved", None)]])], + ); + let report = run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + door(&journal, &["one", "two"], &["one"]), + )) + .expect("the episode runs"); + assert!(report.settled >= 1); + assert_eq!(report.turns, 2, "the parked turn and the one after it"); + let events = journal.events(); + assert!( + events + .iter() + .any(|event| matches!(event, Event::Parked { seat, thread: None } if seat == "one")), + "{events:?}" + ); + assert!( + events + .iter() + .any(|event| matches!(event, Event::Resumed { seat, .. } if seat == "one")), + "{events:?}" + ); + assert!( + !events + .iter() + .any(|event| matches!(event, Event::Nudged { .. })), + "a parked seat is not nudged: {events:?}" + ); + // The host was asked about exactly the seats that were parked. + assert_eq!(*journal.asked.lock().unwrap(), vec![vec!["one".to_owned()]]); +} + +#[test] +fn a_host_that_releases_nobody_ends_the_episode_parked() { + 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![park()]])]); + let ended = run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + door(&journal, &["one", "two"], &["one"]), + )); + assert!( + matches!(&ended, Err(Error::Conduct(tinyhivemind_driver::Error::Parked { seats })) if seats == &["one".to_owned()]), + "{ended:?}" + ); +} + +#[test] +fn the_seats_other_conversations_reach_its_brief_as_context() { + 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(); + // The host keeps one log for both desks, and names them both: the + // episode's own is skipped, marketing is not. + journal + .log + .append_to("marketing", "three", "launch is friday", None, None); + *journal.channels.lock().unwrap() = vec![ + Conversation { + desk_id: "engineering".into(), + desk_name: "Engineering".into(), + thread_root: None, + }, + Conversation { + desk_id: "marketing".into(), + desk_name: "Marketing".into(), + thread_root: None, + }, + ]; + let runner = ScriptRunner::new( + &["one", "two"], + &[("one", vec![vec![complete("done", None)]])], + ); + run(run_episode( + &journal, + &runner, + &driver, + routing, + ConductPolicy::default(), + door(&journal, &["one", "two"], &["one"]), + )) + .expect("the episode runs"); + let prompts = runner.prompts(); + assert_eq!(prompts.len(), 1); + let prompt = &prompts[0].2; + assert!(prompt.contains("## Elsewhere, for context"), "{prompt}"); + assert!( + prompt.contains("### Marketing (marketing)\n@three: launch is friday"), + "{prompt}" + ); + assert!( + !prompt.contains("### Engineering"), + "the turn's own desk is not elsewhere: {prompt}" + ); + assert_ne!(journal.log.latest(), Some(Sequence(0))); +} diff --git a/crates/tinyhivemind-openhuman/src/episode/test/support.rs b/crates/tinyhivemind-openhuman/src/episode/test/support.rs index 6d344755..23c60a1e 100644 --- a/crates/tinyhivemind-openhuman/src/episode/test/support.rs +++ b/crates/tinyhivemind-openhuman/src/episode/test/support.rs @@ -10,14 +10,14 @@ 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::{Conversation, 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 super::super::{Journal, Released}; use crate::MemoryLog; use crate::Result; use crate::runner::{Lane, SeatRunner, TurnJob, TurnResult}; @@ -111,8 +111,9 @@ impl SeatRunner for ScriptRunner { 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()))); + return (seat, lane, TurnResult::Failed("the model went away".into())); } + let parked = calls.iter().any(|(name, _)| *name == "park"); assert!( !calls.iter().any(|(name, _)| *name == "panic"), "the model's task panicked" @@ -120,7 +121,10 @@ impl SeatRunner for ScriptRunner { for (name, arguments) in &calls { let _ = tools.call(&seat, name, arguments); } - (seat, lane, Some(Ok("said".into()))) + if parked { + return (seat, lane, TurnResult::Parked); + } + (seat, lane, TurnResult::Replied("said".into())) }) } } @@ -136,6 +140,12 @@ pub(super) struct TestJournal { pub(super) turns: Mutex>, /// Each desk brief's conversations, by seat, as composed. pub(super) shown: Mutex>, + /// The channels the host names for every seat. + pub(super) channels: Mutex>, + /// Seats the host releases the next time it is asked, then nothing. + pub(super) release: Mutex>>, + /// Every set of parked seats the loop asked about. + pub(super) asked: Mutex>>, } impl TestJournal { @@ -149,6 +159,9 @@ impl TestJournal { events: Mutex::new(Vec::new()), turns: Mutex::new(Vec::new()), shown: Mutex::new(Vec::new()), + channels: Mutex::new(Vec::new()), + release: Mutex::new(VecDeque::new()), + asked: Mutex::new(Vec::new()), } } @@ -165,6 +178,27 @@ impl Journal for TestJournal { &self.log } + fn channels(&self, _seat: &str) -> Vec { + self.channels + .lock() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } + + fn released<'a>(&'a self, parked: &'a [String]) -> Released<'a> { + self.asked + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(parked.to_vec()); + let released = self + .release + .lock() + .unwrap_or_else(PoisonError::into_inner) + .pop_front() + .unwrap_or_default(); + Box::pin(async move { Ok(released) }) + } + fn commit(&self, commit: &Commit) -> Result { Ok(self.log.append( &commit.author, diff --git a/crates/tinyhivemind-openhuman/src/hosted/README.md b/crates/tinyhivemind-openhuman/src/hosted/README.md index 684b2b5e..d4ae8208 100644 --- a/crates/tinyhivemind-openhuman/src/hosted/README.md +++ b/crates/tinyhivemind-openhuman/src/hosted/README.md @@ -24,7 +24,11 @@ the hosted turn's allowlist comes from there (`register_seats`). 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. +`after_turn` runs once a turn has run, with the usage the session +reported, and says what became of it: `Disposition::Done` for a turn that +stands, or `Disposition::Parked` for one that stopped on something only the +host can settle -- an approval it has queued. A parked turn keeps whatever +it called, and the seat is held until the host releases it. It is where a host parks what the turn left waiting on approval, meters the spend, and halts the episode: an error from it is the turn's error, seen by the host loop as any failed turn is, with the calls the turn made before it diff --git a/crates/tinyhivemind-openhuman/src/hosted/mod.rs b/crates/tinyhivemind-openhuman/src/hosted/mod.rs index bddd7cc7..f1ed2631 100644 --- a/crates/tinyhivemind-openhuman/src/hosted/mod.rs +++ b/crates/tinyhivemind-openhuman/src/hosted/mod.rs @@ -47,7 +47,7 @@ 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::runner::{Lane, SeatRunner, TURN_TIMEOUT, TurnJob, TurnResult, unseated}; use crate::{Error, Result}; use admission::Admission; @@ -86,22 +86,35 @@ pub trait EpisodeHost: Journal + 'static { } /// After a turn ran, with its usage when the session reported any. The - /// default does nothing. + /// default meters nothing and lets the turn stand. /// - /// This is where a host parks what the turn left waiting on approval, - /// meters the spend, and decides whether the episode goes on: an error - /// here is the turn's error, and the host loop treats it as it treats - /// any failed turn. + /// This is where a host meters the spend and says what became of the + /// turn: [`Disposition::Done`] for a turn that is what it is, and + /// [`Disposition::Parked`] for one that stopped on something only the + /// host can settle -- an approval it has queued -- which holds the seat + /// until the host releases it. An error here is the turn's error, and + /// the loop treats it as any failed turn. /// /// # Errors /// /// Whatever stops the episode. - fn after_turn(&self, seat: &str, usage: Option<&LastTurnUsage>) -> Result<()> { + fn after_turn(&self, seat: &str, usage: Option<&LastTurnUsage>) -> Result { let _ = (seat, usage); - Ok(()) + Ok(Disposition::Done) } } +/// What the host made of a turn that came back. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum Disposition { + /// Nothing is outstanding: the turn is what it is. + #[default] + Done, + /// The turn stopped on something only the host can settle, and the host + /// has taken it: the seat is held until it says otherwise. + Parked, +} + /// The episode's tools for one seat, and the gate that admits them. pub struct EpisodeBelt { /// The tools, each bound to this seat, each calling the shared record. @@ -322,11 +335,12 @@ impl SeatRunner for HostedRunner { .take(); let finalized = host.after_turn(&seat, last.as_ref()); let result = match (outcome, finalized) { - (Ok(reply), Ok(())) => Ok(reply), - (Ok(_), Err(halt)) => Err(halt), - (Err(error), _) => Err(error), + (Ok(_), Ok(Disposition::Parked)) => TurnResult::Parked, + (Ok(reply), Ok(Disposition::Done)) => TurnResult::Replied(reply), + (Ok(_), Err(halt)) => TurnResult::Failed(halt.to_string()), + (Err(error), _) => TurnResult::Failed(error.to_string()), }; - (seat, lane, Some(result.map_err(|error| error.to_string()))) + (seat, lane, result) }) } } diff --git a/crates/tinyhivemind-openhuman/src/journal/README.md b/crates/tinyhivemind-openhuman/src/journal/README.md index 859818bc..30ef6508 100644 --- a/crates/tinyhivemind-openhuman/src/journal/README.md +++ b/crates/tinyhivemind-openhuman/src/journal/README.md @@ -4,6 +4,8 @@ `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`, +seats. `append` returns the sequence a row was given, and `append_to` puts +one on another desk of the same host, so a seat's other channels can be +read the way a host's log holds them; `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/journal/mod.rs b/crates/tinyhivemind-openhuman/src/journal/mod.rs index e870a492..5ac15033 100644 --- a/crates/tinyhivemind-openhuman/src/journal/mod.rs +++ b/crates/tinyhivemind-openhuman/src/journal/mod.rs @@ -29,6 +29,9 @@ pub struct Row { pub thread: Option, /// On the desk, the one seat it reaches. pub only_for: Option, + /// The desk it is on, when that is not this journal's own: a host's log + /// spans its channels, and a seat reads its others as context. + pub desk: Option, } /// An append-only journal for one desk, held in memory. @@ -62,13 +65,38 @@ impl MemoryLog { self.rows.lock().unwrap_or_else(PoisonError::into_inner) } - /// Append a row and return the sequence it was given. + /// Append a row to this journal's own desk and return the sequence it + /// was given. pub fn append( &self, author: &str, body: &str, thread: Option, only_for: Option<&str>, + ) -> Sequence { + self.append_row(None, author, body, thread, only_for) + } + + /// Append a row to another desk of the same host, sequenced with the + /// rest: one log holds every channel, as a host's does. + pub fn append_to( + &self, + desk: &str, + author: &str, + body: &str, + thread: Option, + only_for: Option<&str>, + ) -> Sequence { + self.append_row(Some(desk), author, body, thread, only_for) + } + + fn append_row( + &self, + desk: Option<&str>, + author: &str, + body: &str, + thread: Option, + only_for: Option<&str>, ) -> Sequence { let mut rows = self.rows(); let sequence = Sequence(rows.last().map_or(self.first, |row| row.sequence.0 + 1)); @@ -78,6 +106,7 @@ impl MemoryLog { body: body.to_owned(), thread, only_for: only_for.map(str::to_owned), + desk: desk.map(str::to_owned), }); sequence } @@ -94,6 +123,7 @@ impl MemoryLog { pub fn desk_since(&self, seat: &str, after: Option) -> Vec { self.rows() .iter() + .filter(|row| row.desk.is_none()) .filter(|row| after.is_none_or(|after| row.sequence > after) && row.thread.is_none()) .filter(|row| { row.only_for @@ -193,7 +223,7 @@ impl SessionLog for MemoryLog { .iter() .map(|row| LogMessage { sequence: row.sequence, - chat_id: Some(self.desk.clone()), + chat_id: Some(row.desk.clone().unwrap_or_else(|| self.desk.clone())), parent: row.thread, author: Self::author(&row.author), content: row.body.clone(), diff --git a/crates/tinyhivemind-openhuman/src/lib.rs b/crates/tinyhivemind-openhuman/src/lib.rs index f4b424d3..2e598a80 100644 --- a/crates/tinyhivemind-openhuman/src/lib.rs +++ b/crates/tinyhivemind-openhuman/src/lib.rs @@ -124,9 +124,9 @@ pub mod raw; pub mod runner; pub use embed::{EmbedRunner, EmbedSeat}; -pub use episode::{Journal, Report, run_episode}; +pub use episode::{Journal, Released, Report, run_episode}; pub use error::{Error, Result}; -pub use hosted::{EpisodeBelt, EpisodeHost, HostedRunner, HostedSeat, HostedTurn}; +pub use hosted::{Disposition, 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/raw/mod.rs b/crates/tinyhivemind-openhuman/src/raw/mod.rs index 2cfd389d..a45369b2 100644 --- a/crates/tinyhivemind-openhuman/src/raw/mod.rs +++ b/crates/tinyhivemind-openhuman/src/raw/mod.rs @@ -55,7 +55,7 @@ use tinyhivemind::Sequence; use tinyhivemind_driver::AgentBinding; use tinyhivemind_tools::EpisodeTools; -use crate::runner::{Lane, SeatRunner, TurnJob, unseated}; +use crate::runner::{Lane, SeatRunner, TurnJob, TurnResult, unseated}; use crate::{Error, Result}; pub use library::LibraryHost; pub use seat::RawSeat; @@ -261,14 +261,14 @@ impl SeatRunner for RawRunner { let contexts = Arc::clone(&self.contexts); Box::pin(async move { let result = match Box::pin(raw_seat.turn(history, &prompt, belt)).await { - Ok(reply) => Some(Ok(reply)), - Err(error) => Some(Err(error.to_string())), + Ok(reply) => TurnResult::Replied(reply), + Err(error) => TurnResult::Failed(error.to_string()), }; - if let Some(Ok(reply)) = &result { + if let Some(reply) = result.reply() { let mut contexts = contexts.lock().unwrap_or_else(PoisonError::into_inner); let context = contexts.entry(seat.clone()).or_default(); context.push(("user".to_owned(), prompt)); - context.push(("assistant".to_owned(), reply.clone())); + context.push(("assistant".to_owned(), reply.to_owned())); } (seat, lane, result) }) diff --git a/crates/tinyhivemind-openhuman/src/runner/README.md b/crates/tinyhivemind-openhuman/src/runner/README.md index db190445..1ff6b913 100644 --- a/crates/tinyhivemind-openhuman/src/runner/README.md +++ b/crates/tinyhivemind-openhuman/src/runner/README.md @@ -4,7 +4,8 @@ The seam a host loop steps a seat through. `SeatRunner::open` registers the turn and the read window, `turn` runs it, `close` drains what was called; open and close are provided, because both runners record into the same `EpisodeTools`. `Lane` is where a turn runs, desk or thread, for the host's -own bookkeeping; `TurnJob` is a running turn. `RunnerKind` names a runner +own bookkeeping; `TurnJob` is a running turn, and `TurnResult` what became +of one -- it replied, it failed, or it parked on the host. `RunnerKind` names a runner and states its one sentence on the mechanics for the standing contract. `test.rs` proves both runners through the seam against the scripted model, diff --git a/crates/tinyhivemind-openhuman/src/runner/mod.rs b/crates/tinyhivemind-openhuman/src/runner/mod.rs index 8e4bb251..4cd9fa64 100644 --- a/crates/tinyhivemind-openhuman/src/runner/mod.rs +++ b/crates/tinyhivemind-openhuman/src/runner/mod.rs @@ -32,8 +32,37 @@ pub enum Lane { Thread(Sequence), } -/// A turn's reply, once it is back: `None` timed out. -pub type TurnResult = Option>; +/// What became of a turn. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum TurnResult { + /// It came back, with what the seat said. What it *called* is in the + /// record, not here. + Replied(String), + /// It did not come back: the model failed, it timed out, or the host + /// stopped it. The string is what a host prints. + Failed(String), + /// It stopped on something only the host can settle -- an approval, + /// typically. Whatever it called first still counts; the seat is held + /// until the host releases it. + Parked, +} + +impl TurnResult { + /// The reply, or nothing for a turn that failed or parked. + #[must_use] + pub fn reply(&self) -> Option<&str> { + match self { + Self::Replied(reply) => Some(reply), + Self::Failed(_) | Self::Parked => None, + } + } + + /// Whether the seat is held on the host. + #[must_use] + pub const fn parked(&self) -> bool { + matches!(self, Self::Parked) + } +} /// The turn a runner returns for a seat it never seated: failed, at once. /// A runner indexes its seats by what the driver proposed, and the driver @@ -43,7 +72,7 @@ pub type TurnResult = Option>; pub fn unseated(seat: String, lane: Lane) -> TurnJob { Box::pin(async move { let failed = format!("`{seat}` is not a seat of this runner"); - (seat, lane, Some(Err(failed))) + (seat, lane, TurnResult::Failed(failed)) }) } diff --git a/crates/tinyhivemind-openhuman/src/runner/test.rs b/crates/tinyhivemind-openhuman/src/runner/test.rs index 1394224a..3b999ceb 100644 --- a/crates/tinyhivemind-openhuman/src/runner/test.rs +++ b/crates/tinyhivemind-openhuman/src/runner/test.rs @@ -17,8 +17,8 @@ use tinyhivemind_tools::{Dispatch, EpisodeTools, SeatEvent, served_specs}; use super::{Lane, RunnerKind, SeatRunner}; use crate::MemoryLog; use crate::{ - EmbedRunner, EpisodeBelt, EpisodeHost, HostedRunner, HostedTurn, Journal, LibraryHost, - RawRunner, Route, offline, register_seats, + Disposition, EmbedRunner, EpisodeBelt, EpisodeHost, HostedRunner, HostedTurn, Journal, + LibraryHost, RawRunner, Route, TurnResult, offline, register_seats, }; use tinyhivemind_driver::{Commit, Note}; @@ -35,6 +35,8 @@ struct TestHost { metered: AtomicBool, /// Whether the hook halts the episode on the next turn. halt: AtomicBool, + /// The next turn stops on the host instead of standing. + park: AtomicBool, } impl Journal for TestHost { @@ -73,7 +75,7 @@ impl EpisodeHost for TestHost { "desk_".into() } - fn after_turn(&self, seat: &str, usage: Option<&LastTurnUsage>) -> crate::Result<()> { + fn after_turn(&self, seat: &str, usage: Option<&LastTurnUsage>) -> crate::Result { assert_eq!(seat, "lead"); self.after.fetch_add(1, Ordering::SeqCst); if usage.is_some() { @@ -84,7 +86,10 @@ impl EpisodeHost for TestHost { "the desk's budget is spent" ))); } - Ok(()) + if self.park.swap(false, Ordering::SeqCst) { + return Ok(Disposition::Parked); + } + Ok(Disposition::Done) } } @@ -144,9 +149,7 @@ async fn one_turn(runner: &R, since: Option) -> (String .await; assert_eq!(seat, "lead"); assert_eq!(lane, Lane::Desk); - let reply = reply - .expect("the turn did not time out") - .expect("the turn ran"); + let reply = reply.reply().map(str::to_owned).expect("the turn ran"); (reply, runner.close("lead")) } @@ -227,7 +230,7 @@ async fn plain(library: LibraryHost) { ) .await; assert_eq!(lane, Lane::Thread(Sequence(1))); - assert!(matches!(outcome, Some(Ok(_))), "{outcome:?}"); + assert!(matches!(outcome, TurnResult::Replied(_)), "{outcome:?}"); assert!( runner.close("lead").is_empty(), "the scripted call names no thread, so the record refused it" @@ -248,6 +251,7 @@ fn hosted(library: LibraryHost, contract: &str) -> (Arc, HostedRunner< after: AtomicUsize::new(0), metered: AtomicBool::new(false), halt: AtomicBool::new(false), + park: AtomicBool::new(false), }); let runner = HostedRunner::seat( Arc::clone(&host), @@ -306,7 +310,7 @@ async fn ghosts(embed: &EmbedRunner, raw: &RawRunner, hosted: &HostedRunner) { ) .await; assert!( - matches!(&halted, Some(Err(error)) if error.contains("budget is spent")), + matches!(&halted, TurnResult::Failed(error) if error.contains("budget is spent")), "{halted:?}" ); assert_eq!( @@ -373,7 +377,7 @@ async fn halts(host: &TestHost, hosted: &HostedRunner) { before + 1, "the hook ran" ); - assert!(matches!(&failed, Some(Err(_))), "{failed:?}"); + assert!(matches!(&failed, TurnResult::Failed(_)), "{failed:?}"); hosted.close("lead"); host.halt.store(false, Ordering::SeqCst); } diff --git a/examples/openhuman/src/bin/conducted/hosted.rs b/examples/openhuman/src/bin/conducted/hosted.rs index c3010cad..0aeb5683 100644 --- a/examples/openhuman/src/bin/conducted/hosted.rs +++ b/examples/openhuman/src/bin/conducted/hosted.rs @@ -89,6 +89,12 @@ impl Journal for DeskJournal { seat, thread: Some(root), } => eprintln!("[nudged] @{seat} in thread {}", root.0), + Event::Parked { seat, thread } => { + eprintln!("[parked] @{seat}{} waits on the operator", place(*thread)); + } + Event::Resumed { seat, thread } => { + eprintln!("[resumed] @{seat}{} was released", place(*thread)); + } Event::Broadcast { seat, to, .. } => { println!("[broadcast] @{seat} -> {}", to.join(", ")); } @@ -158,12 +164,14 @@ impl Journal for DeskJournal { Lane::Thread(root) => format!(" in thread {}", root.0), }; match outcome { - Some(Ok(reply)) => eprintln!( + TurnResult::Replied(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"), + TurnResult::Failed(error) => eprintln!("[turn] @{seat}{where_} failed: {error}"), + TurnResult::Parked => { + eprintln!("[turn] @{seat}{where_} parked: waiting on the desk's operator"); + } } for refusal in refused { eprintln!( @@ -171,12 +179,12 @@ impl Journal for DeskJournal { refusal.tool, refusal.reason ); } - if recorded == 0 { + if recorded == 0 && !outcome.parked() { // 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 { + if let Some(reply) = outcome.reply() { let shown: String = reply.chars().take(REPLY_SHOWN).collect(); let cut = if reply.chars().count() > REPLY_SHOWN { " [...]" @@ -262,3 +270,8 @@ impl EpisodeHost for DeskHost { Box::pin(self.library.scope(turn)) } } + +/// ` in thread N`, or nothing on the desk. +fn place(thread: Option) -> String { + thread.map_or_else(String::new, |root| format!(" in thread {}", root.0)) +} diff --git a/examples/openhuman/src/bin/deepswe_hive.rs b/examples/openhuman/src/bin/deepswe_hive.rs index 6e2b7b32..f5158503 100644 --- a/examples/openhuman/src/bin/deepswe_hive.rs +++ b/examples/openhuman/src/bin/deepswe_hive.rs @@ -16,11 +16,11 @@ use openhuman_embed::{ use serde::{Deserialize, Serialize}; use tinyhivemind::desk::{Desk, ResponderMode}; use tinyhivemind::responder::Probability; -use tinyhivemind_hive::{CompletionEpisodeState, CompletionStep, completion_status}; -use tinyhivemind_openhuman::EmbedSeat; use tinyhivemind_driver::{ - AgentBinding, BroadcastRouting, CommittedUtterance, CompletionDriver, HiveGraph, BoundHive, + AgentBinding, BoundHive, BroadcastRouting, CommittedUtterance, CompletionDriver, HiveGraph, }; +use tinyhivemind_hive::{CompletionEpisodeState, CompletionStep, completion_status}; +use tinyhivemind_openhuman::EmbedSeat; use wiremock::matchers::any; use wiremock::{Mock, MockServer, ResponseTemplate}; diff --git a/examples/openhuman/src/bin/pe1006_hive.rs b/examples/openhuman/src/bin/pe1006_hive.rs index dce212f4..d5c7bd00 100644 --- a/examples/openhuman/src/bin/pe1006_hive.rs +++ b/examples/openhuman/src/bin/pe1006_hive.rs @@ -12,15 +12,15 @@ use openhuman_embed::{ }; use serde_json::json; use tinyhivemind::desk::{Desk, ResponderMode}; +use tinyhivemind_driver::{ + AgentBinding, BoundHive, BroadcastRouting, CommittedUtterance, CompletionDriver, HiveGraph, + HostAction, +}; use tinyhivemind_hive::{ CompletionEpisodeState, CompletionStep, ParticipantCompletion, apply_assignment, completion_status, }; use tinyhivemind_openhuman::EmbedSeat; -use tinyhivemind_driver::{ - AgentBinding, BroadcastRouting, CommittedUtterance, CompletionDriver, HiveGraph, HostAction, - BoundHive, -}; use tinyhivemind_typesafe::JevRouter; use wiremock::matchers::any; use wiremock::{Mock, MockServer, ResponseTemplate}; diff --git a/examples/openhuman/src/main.rs b/examples/openhuman/src/main.rs index 835d284b..80a8bf9b 100644 --- a/examples/openhuman/src/main.rs +++ b/examples/openhuman/src/main.rs @@ -12,11 +12,11 @@ use tinyhivemind::{ desk::{Desk, ResponderMode}, responder::Probability, }; +use tinyhivemind_driver::{AgentBinding, BoundHive, HiveGraph}; use tinyhivemind_embed::{ ConversationKind, ConversationRef, MessageRoute, RouteCandidate, RoutingPolicy, RoutingRequest, }; use tinyhivemind_openhuman::EmbedSeat; -use tinyhivemind_driver::{AgentBinding, HiveGraph, BoundHive}; use tinyhivemind_typesafe::{ ChoiceAnswer, JevRouter, NoulAnswer, SystemOneAnswer, SystemOneRequest, SystemOneResponse, SystemOneTransport, SystemOneTransportFuture, TokenUsage, @@ -192,29 +192,33 @@ async fn run() -> anyhow::Result<()> { vec![ AgentBinding::new( "engineering", - EmbedSeat(runtime.agent( - AgentSpec::new("engineering") - .system_prompt("You are the engineering specialist.") - .config(|config| { - config.agent_registry.entries.push(registry_entry( - "engineering", - "You are the engineering specialist.", - )); - }), - )?), + EmbedSeat( + runtime.agent( + AgentSpec::new("engineering") + .system_prompt("You are the engineering specialist.") + .config(|config| { + config.agent_registry.entries.push(registry_entry( + "engineering", + "You are the engineering specialist.", + )); + }), + )?, + ), ), AgentBinding::new( "legal", - EmbedSeat(runtime.agent( - AgentSpec::new("legal") - .system_prompt("You are the legal specialist.") - .config(|config| { - config - .agent_registry - .entries - .push(registry_entry("legal", "You are the legal specialist.")); - }), - )?), + EmbedSeat( + runtime.agent( + AgentSpec::new("legal") + .system_prompt("You are the legal specialist.") + .config(|config| { + config + .agent_registry + .entries + .push(registry_entry("legal", "You are the legal specialist.")); + }), + )?, + ), ), ], )?;