Park a seat on the host, and brief it on its other conversations

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

crates/tinyhivemind-driver/src/conduct/mod.rs

@@ -124,6 +124,9 @@ pub struct Conductor<'a, A: BoundAgent> {
     shown: BTreeMap<String, usize>,
     /// The assignment each seat was last nudged for on the desk.
     desk_nudged: BTreeMap<String, Sequence>,
+    /// 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<String, Option<Sequence>>,
     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,61 @@ 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<String> {
+        self.parked.keys().cloned().collect()
+    }
+
+    /// A turn stopped on something only the host can settle -- an approval,
+    /// 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<Item = ToolCall>) {
+        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 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);
+        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<Vec<String>> {
         Ok(self
             .driver

crates/tinyhivemind-driver/src/conduct/steps.rs

@@ -134,6 +134,22 @@ pub enum Event {
         /// The thread, or `None` on the desk.
         thread: Option<Sequence>,
     },
+    /// 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<Sequence>,
+    },
+    /// 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<Sequence>,
+    },
     /// A broadcast was placed.
     Broadcast {
         /// The author.

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;

crates/tinyhivemind-driver/src/conduct/test/parked.rs

@@ -0,0 +1,202 @@
+//! 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, broadcast, 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"]));
+}
+
+#[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()]);
+}

crates/tinyhivemind-driver/src/conduct/test/support.rs

@@ -212,6 +212,18 @@ pub(super) fn wave(
     journal: &Journal,
     calls: &[(&str, Vec<Utterance>)],
 ) -> Result<Wave, Error> {
+    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<Utterance>)],
+    parked: &[&str],
+) -> Result<Wave, Error> {
+    // 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);
@@ -227,7 +239,12 @@ pub(super) fn wave(
             .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()? {