Lift the wave loop into the adapter as run_episode

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

@@ -50,7 +50,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use tinyhivemind::speech::{ToolCall, Utterance};
 use tinyhivemind::{Conversation, Sequence};
 use tinyhivemind_embed::RoutingPlan;
-use tinyhivemind_hive::{CompletionEpisodeState, apply_completion};
+use tinyhivemind_hive::CompletionEpisodeState;
 
 use crate::driver::{BroadcastRouting, Channel, ConversationView, EpisodeBrief};
 use crate::{BoundAgent, CompletionDriver, DriverState, Error, Result};
@@ -168,18 +168,24 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
         {
             return Err(Error::UnknownStarter { seat: seat.clone() });
         }
+        // Everyone is assigned at the task's row, and the passed-over seats
+        // are completed on that same row: set directly, because a completion
+        // applied as an event must land strictly above its assignment, and
+        // the task's row may be the log's first, at sequence zero.
         let mut episode = CompletionEpisodeState::opened(
             Conversation {
                 desk_id: door.chat.clone(),
                 desk_name: door.desk_name.clone(),
                 thread_root: None,
             },
-            Sequence(0),
+            door.opened_at,
             door.members.iter().map(String::as_str),
         )?;
-        for id in &door.members {
-            if !door.starters.contains(id) {
-                episode = apply_completion(&episode, id, door.opened_at)?;
+        for participant in &mut episode.participants {
+            if !door.starters.contains(&participant.agent_id) {
+                for record in &mut participant.assignments {
+                    record.completed_at = Some(door.opened_at);
+                }
             }
         }
         let state = driver.start(episode)?;
@@ -301,13 +307,16 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
                 if !taken.insert(seat.clone()) {
                     continue;
                 }
+                // The ask row is the first thing a seat is shown in the
+                // conversation it roots: a first turn there starts just
+                // below it, which for a root at zero is nowhere.
                 let since = child
                     .state
                     .seen()
                     .delivered_through
                     .get(&seat)
                     .copied()
-                    .unwrap_or(child.root);
+                    .or_else(|| child.root.0.checked_sub(1).map(Sequence));
                 turns.push(Turn {
                     channel: Channel::Thread {
                         root: child.root,
@@ -323,13 +332,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
             if !taken.insert(seat.clone()) {
                 continue;
             }
-            let since = self
-                .state
-                .seen()
-                .delivered_through
-                .get(&seat)
-                .copied()
-                .unwrap_or(Sequence(0));
+            let since = self.state.seen().delivered_through.get(&seat).copied();
             turns.push(Turn {
                 seat,
                 channel: Channel::Desk,
@@ -356,21 +359,24 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
     }
 
     /// Open a turn: record that the seat is shown everything through
-    /// `latest`, and that it ran for what it holds, and build its brief.
+    /// `latest` -- the log's newest row, or `None` for a log with none --
+    /// and that it ran for what it holds, and build its brief.
     /// `new_rows` are the rows above [`Turn::since`] in the turn's channel,
     /// rendered by the host; `transcript` is any thread of the desk, whole,
     /// for the conversations the seat is or was in.
     pub fn open_turn(
         &mut self,
         turn: &Turn,
-        latest: Sequence,
+        latest: Option<Sequence>,
         new_rows: Vec<String>,
         mut transcript: impl FnMut(Sequence) -> Vec<String>,
     ) -> EpisodeBrief {
         match turn.channel {
             Channel::Thread { root, .. } => {
                 if let Some(child) = self.children.get_mut(&root) {
-                    child.state.delivered(&turn.seat, latest);
+                    if let Some(latest) = latest {
+                        child.state.delivered(&turn.seat, latest);
+                    }
                     child.state.turn_started(&turn.seat);
                     child.turns += 1;
                     child.turned = true;
@@ -393,7 +399,9 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
                 )
             }
             Channel::Desk => {
-                self.state.delivered(&turn.seat, latest);
+                if let Some(latest) = latest {
+                    self.state.delivered(&turn.seat, latest);
+                }
                 self.state.turn_started(&turn.seat);
                 let views = self.views(&turn.seat, &mut transcript);
                 EpisodeBrief::for_turn(
@@ -408,6 +416,26 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
         }
     }
 
+    /// The conversations `seat` would be shown on its next desk turn, by
+    /// root: those concluded since it last spoke, and any still in progress.
+    /// A host that reads its log asynchronously fetches these transcripts
+    /// before [`open_turn`](Self::open_turn), which reads them by root.
+    #[must_use]
+    pub fn shown_conversations(&self, seat: &str) -> Vec<Sequence> {
+        let cursor = self.shown.get(seat).copied().unwrap_or(0);
+        self.concluded[cursor..]
+            .iter()
+            .filter(|done| done.involves(seat))
+            .map(|done| done.root)
+            .chain(
+                self.children
+                    .values()
+                    .filter(|child| child.involves(seat))
+                    .map(|child| child.root),
+            )
+            .collect()
+    }
+
     /// The conversations a seat is shown on a desk turn: those concluded
     /// since it last spoke, whole, once; and any still in progress.
     fn views(

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

@@ -20,9 +20,13 @@ pub struct Turn {
     pub seat: String,
     /// Where the turn runs: the desk, or a conversation on one of its threads.
     pub channel: Channel,
-    /// The newest row the seat has been shown in this channel: the host
-    /// gives the turn every row above it.
-    pub since: Sequence,
+    /// The newest row the seat has been shown in this channel, or `None`
+    /// for a seat shown nothing there yet: the host gives the turn every
+    /// row above it, which for `None` is every row. A sequence is never
+    /// borrowed to mean "nothing": a host may number its first row zero.
+    /// On the wire the field is present, `null` for `None`.
+    #[serde(deserialize_with = "required_null")]
+    pub since: Option<Sequence>,
 }
 
 impl Turn {
@@ -218,3 +222,14 @@ pub enum Step {
     /// Show this, or don't.
     Event(Event),
 }
+
+/// Deserialize a nullable field that must be present: `serde` fills a
+/// missing `Option` with `None` by default, and a wire form that dropped
+/// the field would then pass as one that sent `null`.
+fn required_null<'de, D, T>(deserializer: D) -> std::result::Result<Option<T>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+    T: serde::Deserialize<'de>,
+{
+    Option::<T>::deserialize(deserializer)
+}

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

@@ -52,7 +52,11 @@ fn an_ask_opens_a_conversation_that_runs_first_and_concludes_to_the_asker() {
         ("two", Some(root)),
         "the askee in the thread runs first; the asker, woken by its own ask row, after"
     );
-    assert_eq!(answered.turns[0].since, root);
+    assert_eq!(
+        answered.turns[0].since,
+        Some(Sequence(root.0 - 1)),
+        "the ask row itself is new to the seat asked"
+    );
     assert!(matches!(
         answered.turns[0].channel,
         Channel::Thread { root: at, ref other, opened_it: false } if at == root && other == "one"
@@ -72,6 +76,8 @@ fn an_ask_opens_a_conversation_that_runs_first_and_concludes_to_the_asker() {
 
     // The asker is released: it runs on the desk, is shown the whole
     // conversation once, and completes.
+    assert_eq!(conductor.shown_conversations("one"), vec![root]);
+    assert!(conductor.shown_conversations("three").is_empty());
     let turns = conductor.turns().expect("turns");
     let brief = conductor.open_turn(&turns[0], journal.latest(), Vec::new(), |root| {
         journal.thread(root)
@@ -287,7 +293,7 @@ fn a_refused_reply_in_a_conversation_is_not_its_answer() {
         .iter()
         .find(|turn| turn.seat == "two")
         .expect("the askee is due");
-    conductor.open_turn(thread_turn, Sequence(1), Vec::new(), |_| Vec::new());
+    conductor.open_turn(thread_turn, Some(Sequence(1)), Vec::new(), |_| Vec::new());
     conductor.record(thread_turn, vec![ToolCall::Speak(complete("too early"))]);
     let mut refused = false;
     while let Some(step) = conductor.step().expect("steps") {

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

@@ -6,6 +6,7 @@ use super::support::{ClarifyRouter, Journal, complete, door, hive, policy, run,
 use crate::conduct::{ConductPolicy, Conductor, starters};
 use crate::driver::BroadcastRouting;
 use crate::{CompletionDriver, Error};
+use tinyhivemind::Sequence;
 use tinyhivemind_embed::{Router, RoutingFallback, RoutingPlan, RoutingRequest};
 
 #[test]
@@ -47,6 +48,37 @@ fn the_door_starts_the_routed_seats_and_completes_the_rest() {
     assert!(conductor.state().quiescent());
 }
 
+#[test]
+fn a_task_at_sequence_zero_opens_the_episode_and_is_new_to_the_starter() {
+    let hive = hive(&["one", "two"]);
+    let driver = CompletionDriver::new(&hive, 4).expect("driver");
+    let route_policy = policy(1);
+    let routing = BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    // A host numbering from zero: the task is row zero, so the passed-over
+    // seat is completed on row zero, which no completion event could land on.
+    let journal = Journal::numbered_from(0);
+    let entrance = door(&["one", "two"], &["one"], &journal);
+    assert_eq!(entrance.opened_at, Sequence(0));
+    let mut conductor =
+        Conductor::open(&driver, routing, ConductPolicy::default(), entrance).expect("opens");
+    let turns = conductor.turns().expect("turns");
+    assert_eq!(seats(&turns), vec![("one", None)]);
+    assert_eq!(
+        turns[0].since, None,
+        "shown nothing yet: row zero is above the watermark, not on it"
+    );
+    let first = wave(&mut conductor, &journal, &[("one", vec![complete("done")])]).expect("wave");
+    assert_eq!(first.turns.len(), 1);
+    assert!(conductor.finished());
+    assert_eq!(journal.bodies(), vec!["the task", "COMPLETE: done"]);
+}
+
 #[test]
 fn the_door_refuses_a_starter_outside_the_desk_and_no_starter_at_all() {
     let hive = hive(&["one", "two"]);

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

@@ -126,12 +126,29 @@ impl Router for ClarifyRouter {
 pub(super) type Row = (Sequence, String, String, Option<Sequence>, Option<String>);
 
 /// The host: rows, and nothing else.
-#[derive(Debug, Default)]
+#[derive(Debug)]
 pub(super) struct Journal {
+    /// The sequence the first row is given.
+    first: u64,
     rows: Mutex<Vec<Row>>,
 }
 
+impl Default for Journal {
+    fn default() -> Self {
+        Self::numbered_from(1)
+    }
+}
+
 impl Journal {
+    /// A journal whose first row is given `first`: some hosts number from
+    /// zero.
+    pub(super) fn numbered_from(first: u64) -> Self {
+        Self {
+            first,
+            rows: Mutex::new(Vec::new()),
+        }
+    }
+
     pub(super) fn append(
         &self,
         author: &str,
@@ -140,17 +157,13 @@ impl Journal {
         only_for: Option<String>,
     ) -> Sequence {
         let mut rows = self.rows.lock().unwrap();
-        let sequence = Sequence(rows.last().map_or(0, |row| row.0.0) + 1);
+        let sequence = Sequence(rows.last().map_or(self.first, |row| row.0.0 + 1));
         rows.push((sequence, author.into(), body.into(), thread, only_for));
         sequence
     }
 
-    pub(super) fn latest(&self) -> Sequence {
-        self.rows
-            .lock()
-            .unwrap()
-            .last()
-            .map_or(Sequence(0), |row| row.0)
+    pub(super) fn latest(&self) -> Option<Sequence> {
+        self.rows.lock().unwrap().last().map(|row| row.0)
     }
 
     pub(super) fn thread(&self, root: Sequence) -> Vec<String> {

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

@@ -41,7 +41,7 @@ fn a_turn_names_its_seat_channel_and_watermark() {
             other: "one".into(),
             opened_it: false,
         },
-        since: Sequence(4),
+        since: Some(Sequence(4)),
     };
     let wire = serde_json::to_value(&thread).expect("serializes");
     assert_eq!(
@@ -58,11 +58,14 @@ fn a_turn_names_its_seat_channel_and_watermark() {
     let desk = Turn {
         seat: "one".into(),
         channel: Channel::Desk,
-        since: Sequence(0),
+        since: None,
     };
+    let wire = serde_json::to_value(&desk).expect("serializes");
+    assert_eq!(wire["channel"], json!({"kind": "desk"}));
     assert_eq!(
-        serde_json::to_value(&desk).expect("serializes")["channel"],
-        json!({"kind": "desk"})
+        wire["since"],
+        json!(null),
+        "nothing shown yet is null on the wire, never a sequence"
     );
     round_trips(&desk);
 }

crates/tinyhivemind-openhuman/Cargo.toml

@@ -49,7 +49,7 @@ serde_json.workspace = true
 thiserror.workspace = true
 # A turn has a wall; `tokio::time::timeout` is it. Both harnesses run on
 # tokio already.
-tokio = { workspace = true, features = ["sync", "time"] }
+tokio = { workspace = true, features = ["rt", "sync", "time"] }
 tinyhivemind.workspace = true
 # The record every call lands in, and the served definitions.
 tinyhivemind-tools.workspace = true
@@ -61,6 +61,8 @@ wiremock = { workspace = true, optional = true }
 
 [dev-dependencies]
 tempfile.workspace = true
+# The routing surfaces, for a test that builds its own hive.
+tinyhivemind-embed.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "time"] }
 wiremock.workspace = true
 

crates/tinyhivemind-openhuman/README.md

@@ -10,7 +10,10 @@ This crate is the host's side of that seam for OpenHuman, three ways:
 | `EmbedRunner` | an `openhuman-embed` `AgentSpec` agent on a runtime the host booted | the three MCP dispatchers, dialling `tinyhivemind-mcp`'s server | OpenHuman's own session, stable for the episode |
 | `RawRunner` | an `OpenHumanSessionHost` built one level down, per turn | the same tools in-process, each calling `EpisodeTools::call` | a per-seat log this crate seeds the next session with |
 
-All three implement `SeatRunner`, the seam: open a turn, run it, close it and
+`run_episode` runs one episode from its door to quiescence over any of them
+and a `Journal` the host implements -- its log, and how it appends the
+conductor's rows -- so a host builds a driver, a door and a runner and calls
+one function. All three implement `SeatRunner`, the seam: open a turn, run it, close it and
 take what was called. Open and close are the same for every runner, because
 every call lands in the same `EpisodeTools`, so the driver drains identical
 events and a seat is refused and acknowledged in the same words whichever

crates/tinyhivemind-openhuman/src/README.md

@@ -2,10 +2,12 @@
 
 | Path | Purpose |
 |---|---|
-| `lib.rs` | Crate overview and the public surface: `SeatRunner`, `RunnerKind`, `HostedRunner`, `EpisodeHost`, `EpisodeBelt`, `EmbedRunner`, `EmbedSeat`, `RawRunner`, `RawSeat`, `LibraryHost`, `Route`, `register_seats`, `offline`. |
-| `error/` | What seating or running a seat can fail with. |
+| `lib.rs` | Crate overview and the public surface: `run_episode`, `Journal`, `Report`, `SeatRunner`, `RunnerKind`, `HostedRunner`, `EpisodeHost`, `EpisodeBelt`, `EmbedRunner`, `EmbedSeat`, `RawRunner`, `RawSeat`, `LibraryHost`, `Route`, `register_seats`, `offline`. |
+| `error/` | What seating or running a seat, or an episode, can fail with. |
+| `episode/` | `run_episode` over a `Journal`: one episode from its door to quiescence. |
 | `runner/` | The seam: open, run, close; `Lane`, `TurnJob`; which runner the environment names. |
+| `journal/` | `MemoryLog`, an in-memory journal that is a real `SessionLog`; always compiled. |
 | `hosted/` | Seats as the host's own agents, built through `EpisodeHost`, seeded from the host's log. |
 | `embed/` | Seats as `openhuman-embed` agents, tools over MCP. |
 | `raw/` | Seats as raw sessions, tools in-process: the belt, the gate, the memory that keeps nothing. |
-| `offline/` | The scripted model, the backend stub, the offline config and an in-memory journal that is a real `SessionLog`, behind the `offline` feature and in tests. |
+| `offline/` | The scripted model, the backend stub and the offline config, behind the `offline` feature and in tests. |

crates/tinyhivemind-openhuman/src/embed/mod.rs

@@ -120,7 +120,7 @@ impl SeatRunner for EmbedRunner {
 
     /// One session per seat for the whole episode, so `OpenHuman` appends to the
     /// context the agent already holds rather than rebuilding one.
-    fn turn(&self, seat: String, lane: Lane, _since: Sequence, prompt: String) -> TurnJob {
+    fn turn(&self, seat: String, lane: Lane, _since: Option<Sequence>, prompt: String) -> TurnJob {
         let Some(agent) = self.agents.get(&seat).cloned() else {
             return unseated(seat, lane);
         };