From c9fc0a9e09611164dd450566aa2e20c564e0b377 Mon Sep 17 00:00:00 2001
From: sanil-23 <sanil@alphahuman.xyz>
Date: Wed, 23 Sep 2026 11:52:02 +0530
Subject: [PATCH 1/4] Let an episode survive a restart

A host that checkpoints the driver's state could resume the episode but
not the conductor: the open conversations, the per-seat watermarks, the
nudge counts, the concluded list and who is parked all lived only in
memory, so a restart lost them and the room started over.

`Conductor::snapshot` answers `Some` only between waves -- every row
committed, nothing in flight, no commit outstanding -- because a snapshot
mid-wave would either lose the rows the host has not appended yet or
duplicate them on resume. `Conductor::resume` rebuilds from one, on a
driver and a routing the host supplies again: a live router is not state,
and a policy the operator changed between restarts should be the new one.

The loop checkpoints once per settled wave through `Journal::checkpoint`,
and `resume_episode` carries an episode on from a snapshot.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../tinyhivemind-driver/src/conduct/README.md |   5 +
 .../tinyhivemind-driver/src/conduct/child.rs  |  10 +-
 crates/tinyhivemind-driver/src/conduct/mod.rs | 109 ++++++++++++++++++
 .../src/conduct/test/parked.rs                |  99 ++++++++++++++++
 .../tinyhivemind-driver/src/conduct/wave.rs   |  10 ++
 crates/tinyhivemind-driver/src/lib.rs         |   3 +-
 .../src/episode/README.md                     |   5 +
 .../tinyhivemind-openhuman/src/episode/mod.rs |  76 +++++++++++-
 .../src/episode/test/support.rs               |  12 ++
 .../src/episode/test/watermark.rs             |  63 +++++++++-
 crates/tinyhivemind-openhuman/src/lib.rs      |   2 +-
 11 files changed, 385 insertions(+), 9 deletions(-)

diff --git a/crates/tinyhivemind-driver/src/conduct/README.md b/crates/tinyhivemind-driver/src/conduct/README.md
index c9db27c3..930e06e8 100644
--- a/crates/tinyhivemind-driver/src/conduct/README.md
+++ b/crates/tinyhivemind-driver/src/conduct/README.md
@@ -31,6 +31,11 @@ The rules, each with the decision it comes from:
   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.
+- **Checkpointing**: `snapshot` answers `Some` only between waves -- every
+  row committed, nothing in flight, no commit outstanding -- and `resume`
+  rebuilds a conductor from one, on a driver and a routing the host supplies
+  again. Mid-wave it answers `None`: a snapshot there would either lose the
+  rows the host has not appended or duplicate them on resume.
 - **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/child.rs b/crates/tinyhivemind-driver/src/conduct/child.rs
index 5265da5f..e19186d4 100644
--- a/crates/tinyhivemind-driver/src/conduct/child.rs
+++ b/crates/tinyhivemind-driver/src/conduct/child.rs
@@ -1,5 +1,6 @@
 //! One open conversation: a thread of the desk, run as its own episode.
 
+use serde::{Deserialize, Serialize};
 use tinyhivemind::Sequence;
 
 use crate::driver::{ConversationView, DriverState};
@@ -8,7 +9,7 @@ use crate::driver::{ConversationView, DriverState};
 /// with the asker recorded here (ADR 0023). One question, one answer: the
 /// seat asked concludes with `complete_episode`, and its message is the
 /// answer; a follow-up is a further ask.
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub(super) struct Child {
     pub(super) root: Sequence,
     pub(super) asker: String,
@@ -20,7 +21,10 @@ pub(super) struct Child {
     pub(super) last_by_askee: Option<String>,
     /// Whether the seat asked has been told once that it has not answered.
     pub(super) nudged: bool,
-    /// Whether the seat asked took a turn in this wave.
+    /// Whether the seat asked took a turn in this wave. Within-wave
+    /// bookkeeping: a snapshot is only taken between waves, where it is
+    /// false, so it is not carried across one.
+    #[serde(skip)]
     pub(super) turned: bool,
 }
 
@@ -82,7 +86,7 @@ impl Child {
 
 /// A conversation that concluded, kept for the context of the seats that had
 /// it: shown whole once to each, on its next desk turn.
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
 pub(super) struct Concluded {
     pub(super) root: Sequence,
     pub(super) asker: String,
diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs
index c3799e7b..815f77b7 100644
--- a/crates/tinyhivemind-driver/src/conduct/mod.rs
+++ b/crates/tinyhivemind-driver/src/conduct/mod.rs
@@ -47,6 +47,8 @@ mod wave;
 
 use std::collections::{BTreeMap, BTreeSet};
 
+use serde::{Deserialize, Serialize};
+
 use tinyhivemind::speech::{ToolCall, Utterance};
 use tinyhivemind::{Conversation, Sequence};
 use tinyhivemind_embed::RoutingPlan;
@@ -110,6 +112,46 @@ pub fn starters(plan: &RoutingPlan, fallback: &str) -> Vec<String> {
     }
 }
 
+/// Everything a conductor needs to be rebuilt: the episode as the driver
+/// folds it, the conversations open and concluded, and what each seat has
+/// been shown, nudged for, or held on.
+///
+/// Taken between waves and nowhere else. Mid-wave a conductor holds rows the
+/// host has not appended yet and a commit whose sequence it has not been
+/// told; a snapshot there would either lose those rows or duplicate them on
+/// resume, so [`Conductor::snapshot`] answers `None` until the wave settles.
+///
+/// The driver, the routing and the policy are **not** here. They are the
+/// host's to supply again on resume, exactly as they were on open: a router
+/// is a live object, and a policy the operator changed between restarts
+/// should be the new one.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub struct ConductorState {
+    /// The desk's id, as every tool call names it.
+    pub chat: String,
+    /// The desk's display name.
+    pub desk_name: String,
+    /// The episode the driver folds.
+    pub state: DriverState,
+    /// The conversations still open, by their ask row.
+    children: Vec<(Sequence, Child)>,
+    /// The conversations that concluded, oldest first.
+    concluded: Vec<Concluded>,
+    /// How many concluded conversations each seat has been shown.
+    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.
+    parked: BTreeMap<String, Option<Sequence>>,
+    /// Turns run so far.
+    turns: u64,
+    /// Waves proposed so far.
+    waves: u64,
+    /// Seats completed with their work for a spent broadcast budget.
+    discharged: u64,
+}
+
 /// The desk episode, its conversations, and the rules between them.
 pub struct Conductor<'a, A: BoundAgent> {
     driver: &'a CompletionDriver<'a, A>,
@@ -358,6 +400,73 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
         Ok(turns)
     }
 
+    /// Everything needed to rebuild this conductor, or `None` mid-wave.
+    ///
+    /// A host checkpoints where the loop already pauses: after
+    /// [`step`](Self::step) has returned `None` and before the next wave
+    /// begins. Anywhere else there are rows in flight, and the snapshot
+    /// would be a lie about what the journal holds.
+    #[must_use]
+    pub fn snapshot(&self) -> Option<ConductorState> {
+        if !self.wave.settled() {
+            return None;
+        }
+        Some(ConductorState {
+            chat: self.chat.clone(),
+            desk_name: self.desk_name.clone(),
+            state: self.state.clone(),
+            children: self
+                .children
+                .iter()
+                .map(|(root, child)| (*root, child.clone()))
+                .collect(),
+            concluded: self.concluded.clone(),
+            shown: self.shown.clone(),
+            desk_nudged: self.desk_nudged.clone(),
+            parked: self.parked.clone(),
+            turns: self.turns,
+            waves: self.waves,
+            discharged: self.discharged,
+        })
+    }
+
+    /// Rebuild a conductor from a snapshot, on a driver and a routing the
+    /// host supplies again.
+    ///
+    /// The episode carries on where it paused: the same conversations are
+    /// open, the same seats are held, and a seat already nudged for its
+    /// assignment is not nudged again for it.
+    ///
+    /// # Errors
+    ///
+    /// The driver refusing the episode -- a state naming a seat or a desk
+    /// this hive does not have.
+    pub fn resume(
+        driver: &'a CompletionDriver<'a, A>,
+        routing: BroadcastRouting<'a>,
+        policy: ConductPolicy,
+        snapshot: ConductorState,
+    ) -> Result<Self> {
+        let state = driver.resume(snapshot.state)?;
+        Ok(Self {
+            driver,
+            routing,
+            chat: snapshot.chat,
+            desk_name: snapshot.desk_name,
+            policy,
+            state,
+            children: snapshot.children.into_iter().collect(),
+            concluded: snapshot.concluded,
+            shown: snapshot.shown,
+            desk_nudged: snapshot.desk_nudged,
+            parked: snapshot.parked,
+            turns: snapshot.turns,
+            waves: snapshot.waves,
+            discharged: snapshot.discharged,
+            wave: Wave::default(),
+        })
+    }
+
     /// 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.
diff --git a/crates/tinyhivemind-driver/src/conduct/test/parked.rs b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
index 8e891f21..88ec6019 100644
--- a/crates/tinyhivemind-driver/src/conduct/test/parked.rs
+++ b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
@@ -200,3 +200,102 @@ fn what_a_seat_said_before_it_parked_is_recorded() {
     );
     assert_eq!(conductor.parked(), vec!["one".to_owned()]);
 }
+
+#[test]
+fn a_snapshot_is_taken_between_waves_and_carries_the_episode_on() {
+    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 asks two, and two parks: a conversation open and a seat held is
+    // the most state a snapshot has to carry.
+    wave(
+        &mut conductor,
+        &journal,
+        &[("one", vec![ask("two", "may I ship?")])],
+    )
+    .expect("wave");
+    wave_parking(&mut conductor, &journal, &[], &["two"]).expect("wave");
+    let root = Sequence(2);
+    let snapshot = conductor.snapshot().expect("the wave settled");
+    assert_eq!(snapshot.chat, "engineering");
+    assert_eq!(conductor.parked(), vec!["two".to_owned()]);
+    let turns_before = conductor.turns_run();
+
+    // Through the wire and back, onto a driver the host supplies again.
+    let wire = serde_json::to_string(&snapshot).expect("serializes");
+    let restored: crate::ConductorState = serde_json::from_str(&wire).expect("deserializes");
+    let routing = BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    let mut resumed =
+        Conductor::resume(&driver, routing, ConductPolicy::default(), restored).expect("resumes");
+    assert_eq!(resumed.turns_run(), turns_before, "the count carries");
+    assert_eq!(
+        resumed.parked(),
+        vec!["two".to_owned()],
+        "the held seat is still held"
+    );
+    // Released, it answers in the conversation that was open before the
+    // restart, and the episode concludes it.
+    resumed.resume_seat("two");
+    let answered = wave(
+        &mut resumed,
+        &journal,
+        &[("two", vec![complete("ship it")])],
+    )
+    .expect("wave");
+    assert_eq!(seats(&answered.turns)[0], ("two", Some(root)));
+    assert_eq!(
+        resumed.conversations(),
+        1,
+        "the conversation survived the restart and concluded"
+    );
+}
+
+#[test]
+fn a_snapshot_is_refused_while_a_commit_is_in_flight() {
+    let hive = hive(&["one", "two"]);
+    let driver = CompletionDriver::new(&hive, 4).expect("driver");
+    let route_policy = policy(1);
+    let routing = BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    let journal = Journal::default();
+    let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal);
+    assert!(
+        conductor.snapshot().is_some(),
+        "a fresh conductor is between waves"
+    );
+    for step in conductor.begin_wave() {
+        let _ = step;
+    }
+    let turns = conductor.turns().expect("turns");
+    conductor.record(
+        &turns[0],
+        [tinyhivemind::speech::ToolCall::Speak(complete("done"))],
+    );
+    // A commit is queued and the host has not been told its sequence.
+    let step = conductor.step().expect("a step");
+    assert!(matches!(step, Some(crate::Step::Commit(_))));
+    assert!(
+        conductor.snapshot().is_none(),
+        "mid-wave there are rows the journal does not hold yet"
+    );
+}
diff --git a/crates/tinyhivemind-driver/src/conduct/wave.rs b/crates/tinyhivemind-driver/src/conduct/wave.rs
index c680f121..588a418c 100644
--- a/crates/tinyhivemind-driver/src/conduct/wave.rs
+++ b/crates/tinyhivemind-driver/src/conduct/wave.rs
@@ -59,6 +59,16 @@ pub(super) struct Wave {
 }
 
 impl Wave {
+    /// Nothing is in flight: no step queued, no commit queued, no commit
+    /// outstanding, and the phase machine is idle. The one point a snapshot
+    /// describes the journal truthfully.
+    pub(super) fn settled(&self) -> bool {
+        matches!(self.phase, Phase::Idle)
+            && self.steps.is_empty()
+            && self.commits.is_empty()
+            && self.outstanding.is_none()
+    }
+
     pub(super) fn begin(&mut self, nothing_due: bool) {
         self.phase = Phase::Threads;
         self.force_conclusions = nothing_due;
diff --git a/crates/tinyhivemind-driver/src/lib.rs b/crates/tinyhivemind-driver/src/lib.rs
index 1ae94c6b..4136c298 100644
--- a/crates/tinyhivemind-driver/src/lib.rs
+++ b/crates/tinyhivemind-driver/src/lib.rs
@@ -78,7 +78,8 @@ pub mod graph;
 mod test_support;
 
 pub use conduct::{
-    Commit, ConductPolicy, Conductor, Door, Event, Note, Refusal, Step, Turn, starters,
+    Commit, ConductPolicy, Conductor, ConductorState, Door, Event, Note, Refusal, Step, Turn,
+    starters,
 };
 pub use driver::{
     AssignmentSpend, BroadcastRouting, Channel, CommittedUtterance, CompletionDriver,
diff --git a/crates/tinyhivemind-openhuman/src/episode/README.md b/crates/tinyhivemind-openhuman/src/episode/README.md
index c69c9f6f..2847183b 100644
--- a/crates/tinyhivemind-openhuman/src/episode/README.md
+++ b/crates/tinyhivemind-openhuman/src/episode/README.md
@@ -21,6 +21,11 @@ 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.
 
+`checkpoint` is handed a `ConductorState` once per wave, after the wave
+settles, and `resume_episode` carries an episode on from one: the same
+conversations open, the same seats held. A host that keeps nothing loses a
+running episode to a restart.
+
 `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
diff --git a/crates/tinyhivemind-openhuman/src/episode/mod.rs b/crates/tinyhivemind-openhuman/src/episode/mod.rs
index 24d0b31e..1111918f 100644
--- a/crates/tinyhivemind-openhuman/src/episode/mod.rs
+++ b/crates/tinyhivemind-openhuman/src/episode/mod.rs
@@ -28,8 +28,8 @@ use tinyhivemind::{
     SessionMessage, SessionQuery, gather_elsewhere, project_session,
 };
 use tinyhivemind_driver::{
-    BoundAgent, BroadcastRouting, Commit, CompletionDriver, ConductPolicy, Conductor, Door,
-    ElsewhereView, EpisodeBrief, Event, Note, Step, Turn,
+    BoundAgent, BroadcastRouting, Commit, CompletionDriver, ConductPolicy, Conductor,
+    ConductorState, Door, ElsewhereView, EpisodeBrief, Event, Note, Step, Turn,
 };
 use tinyhivemind_tools::{Dispatch, Refusal};
 
@@ -80,6 +80,23 @@ pub trait Journal: Send + Sync {
         Vec::new()
     }
 
+    /// The episode paused where it can be resumed: every row it committed
+    /// is in the journal and nothing is in flight. The default keeps
+    /// nothing, and such a host loses a running episode to a restart.
+    ///
+    /// Called once per wave, after the wave settles. A host stores the
+    /// snapshot beside its rows and hands it to
+    /// [`resume_episode`] on boot.
+    ///
+    /// # Errors
+    ///
+    /// The host failing to store it, which ends the episode: an episode
+    /// that cannot be checkpointed is one a restart would lose silently.
+    fn checkpoint(&self, state: &ConductorState) -> Result<()> {
+        let _ = state;
+        Ok(())
+    }
+
     /// Nothing is due and these seats are parked: the seats the host has
     /// settled, waiting until it has one.
     ///
@@ -161,7 +178,55 @@ where
         desk_name: door.desk_name.clone(),
         thread_root: None,
     };
-    let mut conductor = Conductor::open(driver, routing, policy, door)?;
+    let conductor = Conductor::open(driver, routing, policy, door)?;
+    drive(journal, runner, conductor, desk).await
+}
+
+/// Carry on an episode from a snapshot the host stored.
+///
+/// The same loop as [`run_episode`], opened from
+/// [`Conductor::resume`](tinyhivemind_driver::Conductor::resume) rather than
+/// from a door: the same conversations are open, the same seats are held,
+/// and the rows already committed are already in the host's journal.
+///
+/// # Errors
+///
+/// Whatever [`run_episode`] errors on, plus the driver refusing the snapshot
+/// -- one naming a seat or a desk this hive does not have.
+pub async fn resume_episode<A, J, R>(
+    journal: &J,
+    runner: &R,
+    driver: &CompletionDriver<'_, A>,
+    routing: BroadcastRouting<'_>,
+    policy: ConductPolicy,
+    snapshot: ConductorState,
+) -> Result<Report>
+where
+    A: BoundAgent,
+    J: Journal,
+    R: SeatRunner,
+{
+    let desk = Conversation {
+        desk_id: snapshot.chat.clone(),
+        desk_name: snapshot.desk_name.clone(),
+        thread_root: None,
+    };
+    let conductor = Conductor::resume(driver, routing, policy, snapshot)?;
+    drive(journal, runner, conductor, desk).await
+}
+
+/// The loop itself, however the conductor was opened.
+async fn drive<A, J, R>(
+    journal: &J,
+    runner: &R,
+    mut conductor: Conductor<'_, A>,
+    desk: Conversation,
+) -> Result<Report>
+where
+    A: BoundAgent,
+    J: Journal,
+    R: SeatRunner,
+{
     loop {
         if conductor.finished() {
             break;
@@ -209,6 +274,11 @@ where
         while let Some(step) = conductor.step()? {
             settle(journal, &mut conductor, step).await?;
         }
+        // The wave settled: every row it produced is in the journal and
+        // nothing is in flight, which is the one point a snapshot is true.
+        if let Some(snapshot) = conductor.snapshot() {
+            journal.checkpoint(&snapshot)?;
+        }
     }
     Ok(Report {
         turns: conductor.turns_run(),
diff --git a/crates/tinyhivemind-openhuman/src/episode/test/support.rs b/crates/tinyhivemind-openhuman/src/episode/test/support.rs
index 23c60a1e..18c22089 100644
--- a/crates/tinyhivemind-openhuman/src/episode/test/support.rs
+++ b/crates/tinyhivemind-openhuman/src/episode/test/support.rs
@@ -11,6 +11,7 @@ use serde_json::{Value, json};
 use tinyhivemind::desk::{Desk, ResponderMode};
 use tinyhivemind::responder::Probability;
 use tinyhivemind::{Conversation, Sequence, SessionFuture, SessionLog};
+use tinyhivemind_driver::ConductorState;
 use tinyhivemind_driver::{
     AgentBinding, BoundAgent, BoundHive, Commit, Door, EpisodeBrief, Event, HiveGraph, Note,
 };
@@ -146,6 +147,8 @@ pub(super) struct TestJournal {
     pub(super) release: Mutex<VecDeque<Vec<String>>>,
     /// Every set of parked seats the loop asked about.
     pub(super) asked: Mutex<Vec<Vec<String>>>,
+    /// Every snapshot the loop handed over, in order.
+    pub(super) checkpoints: Mutex<Vec<ConductorState>>,
 }
 
 impl TestJournal {
@@ -162,6 +165,7 @@ impl TestJournal {
             channels: Mutex::new(Vec::new()),
             release: Mutex::new(VecDeque::new()),
             asked: Mutex::new(Vec::new()),
+            checkpoints: Mutex::new(Vec::new()),
         }
     }
 
@@ -178,6 +182,14 @@ impl Journal for TestJournal {
         &self.log
     }
 
+    fn checkpoint(&self, state: &ConductorState) -> Result<()> {
+        self.checkpoints
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .push(state.clone());
+        Ok(())
+    }
+
     fn channels(&self, _seat: &str) -> Vec<Conversation> {
         self.channels
             .lock()
diff --git a/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs b/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs
index bb19444e..7eec4cdb 100644
--- a/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs
+++ b/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs
@@ -8,7 +8,9 @@ use std::sync::atomic::AtomicBool;
 use tinyhivemind::Sequence;
 use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Door};
 
-use super::super::run_episode;
+use super::super::{resume_episode, run_episode};
+use serde_json::json;
+
 use super::support::{GrowingLog, ScriptRunner, TestJournal, complete, door, hive, policy, run};
 use crate::journal::MemoryLog;
 
@@ -106,3 +108,62 @@ fn a_row_the_host_appends_above_the_wave_watermark_is_shown_once_and_later() {
     assert!(prompts[1].2.contains("one more thing"), "{}", prompts[1].2);
     assert_eq!(runner.since(), vec![None, Some(opened_at)]);
 }
+
+#[test]
+fn an_episode_is_checkpointed_every_wave_and_carries_on_from_one() {
+    let hive = hive(&["one", "two"]);
+    let driver = CompletionDriver::new(&hive, 4).expect("driver");
+    let route_policy = policy();
+    let routing = || BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    let journal = TestJournal::new();
+    // One parks on its first turn; with nobody released the episode ends
+    // parked, leaving a checkpoint behind with the seat still held.
+    let runner = ScriptRunner::new(&["one", "two"], &[("one", vec![vec![("park", json!({}))]])]);
+    let entrance = door(&journal, &["one", "two"], &["one"]);
+    let stopped = run(run_episode(
+        &journal,
+        &runner,
+        &driver,
+        routing(),
+        ConductPolicy::default(),
+        entrance,
+    ));
+    assert!(stopped.is_err(), "{stopped:?}");
+    let snapshot = journal
+        .checkpoints
+        .lock()
+        .unwrap()
+        .last()
+        .cloned()
+        .expect("a wave settled before the episode stopped");
+    assert_eq!(snapshot.chat, "engineering");
+
+    // A fresh process: the same journal, a new runner, resumed from the
+    // snapshot. The seat is still held, and released it completes.
+    let after = ScriptRunner::new(
+        &["one", "two"],
+        &[("one", vec![vec![complete("approved", None)]])],
+    );
+    let report = run(resume_episode(
+        &journal,
+        &after,
+        &driver,
+        routing(),
+        ConductPolicy::default(),
+        snapshot,
+    ));
+    // Nobody released it, so it stops parked again rather than silently
+    // starting over: the hold survived the restart, which is the point.
+    assert!(report.is_err(), "{report:?}");
+    assert!(
+        after.prompts().is_empty(),
+        "a held seat is not proposed on resume: {:?}",
+        after.prompts()
+    );
+}
diff --git a/crates/tinyhivemind-openhuman/src/lib.rs b/crates/tinyhivemind-openhuman/src/lib.rs
index 2e598a80..f42ac2f5 100644
--- a/crates/tinyhivemind-openhuman/src/lib.rs
+++ b/crates/tinyhivemind-openhuman/src/lib.rs
@@ -124,7 +124,7 @@ pub mod raw;
 pub mod runner;
 
 pub use embed::{EmbedRunner, EmbedSeat};
-pub use episode::{Journal, Released, Report, run_episode};
+pub use episode::{Journal, Released, Report, resume_episode, run_episode};
 pub use error::{Error, Result};
 pub use hosted::{Disposition, EpisodeBelt, EpisodeHost, HostedRunner, HostedSeat, HostedTurn};
 pub use journal::MemoryLog;

From e029eb9374985000e7995e3d1d48118e611ff363 Mon Sep 17 00:00:00 2001
From: sanil-23 <sanil@alphahuman.xyz>
Date: Wed, 23 Sep 2026 12:16:05 +0530
Subject: [PATCH 2/4] Bound a crash to one row, not a whole wave

The checkpoint was written after the wave settled, so every row the wave
had already committed was durable while the snapshot describing them was
not. A crash in that window left the journal ahead of the state: the
resumed conductor had folded none of those commits, re-proposed the seats
that wrote them, and the desk ended with two completion rows for one
assignment.

`ConductorState` now carries the wave in progress -- what has been said
and not yet committed, and the steps the host has not taken -- so a
snapshot is exact rather than per-wave and a host checkpoints after every
committed row. The residual window is that one row, and it is documented
where a host will read it. `snapshot` answers `None` only while the host
holds a commit whose sequence it has not reported, which is the one thing
the conductor genuinely cannot know.

A wave that commits nothing -- one that only parked or nudged a seat --
is checkpointed at its end, so the hold survives a restart too. Resuming
drains the restored wave before proposing a new one, because
`begin_wave` resets the phase. `Child::turned` is carried rather than
skipped, now that a wave can come back mid-flight and the nudge owed to a
silent askee rides on it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../tinyhivemind-driver/src/conduct/README.md | 14 +++--
 .../tinyhivemind-driver/src/conduct/child.rs  |  7 +--
 crates/tinyhivemind-driver/src/conduct/mod.rs | 56 ++++++++++++++-----
 .../src/conduct/test/parked.rs                | 23 ++++++--
 .../tinyhivemind-driver/src/conduct/wave.rs   | 28 ++++++++--
 .../src/episode/README.md                     | 15 +++--
 .../tinyhivemind-openhuman/src/episode/mod.rs | 49 ++++++++++++----
 .../src/episode/test/watermark.rs             | 50 ++++++++++++++++-
 8 files changed, 194 insertions(+), 48 deletions(-)

diff --git a/crates/tinyhivemind-driver/src/conduct/README.md b/crates/tinyhivemind-driver/src/conduct/README.md
index 930e06e8..419c950d 100644
--- a/crates/tinyhivemind-driver/src/conduct/README.md
+++ b/crates/tinyhivemind-driver/src/conduct/README.md
@@ -31,11 +31,15 @@ The rules, each with the decision it comes from:
   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.
-- **Checkpointing**: `snapshot` answers `Some` only between waves -- every
-  row committed, nothing in flight, no commit outstanding -- and `resume`
-  rebuilds a conductor from one, on a driver and a routing the host supplies
-  again. Mid-wave it answers `None`: a snapshot there would either lose the
-  rows the host has not appended or duplicate them on resume.
+- **Checkpointing**: `snapshot` carries the wave in progress as well as the
+  episode, so a host checkpoints after **every committed row** rather than
+  once per wave, and a crash replays at most the one row whose sequence had
+  not been reported yet. It answers `None` only while the host holds a
+  commit it has not reported through `committed`: the conductor cannot say
+  whether that row landed, so it writes no claim either way. `resume`
+  rebuilds from a snapshot on a driver and a routing the host supplies
+  again, and a caller drains `step` before proposing a new wave -- a
+  restored wave is dropped otherwise, because `begin_wave` resets the phase.
 - **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/child.rs b/crates/tinyhivemind-driver/src/conduct/child.rs
index e19186d4..9159c95a 100644
--- a/crates/tinyhivemind-driver/src/conduct/child.rs
+++ b/crates/tinyhivemind-driver/src/conduct/child.rs
@@ -21,10 +21,9 @@ pub(super) struct Child {
     pub(super) last_by_askee: Option<String>,
     /// Whether the seat asked has been told once that it has not answered.
     pub(super) nudged: bool,
-    /// Whether the seat asked took a turn in this wave. Within-wave
-    /// bookkeeping: a snapshot is only taken between waves, where it is
-    /// false, so it is not carried across one.
-    #[serde(skip)]
+    /// Whether the seat asked took a turn in this wave. Carried across a
+    /// snapshot: a wave can be resumed mid-flight, and dropping this would
+    /// lose the nudge owed to a seat that was asked and said nothing.
     pub(super) turned: bool,
 }
 
diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs
index 815f77b7..a90dd3bf 100644
--- a/crates/tinyhivemind-driver/src/conduct/mod.rs
+++ b/crates/tinyhivemind-driver/src/conduct/mod.rs
@@ -116,10 +116,12 @@ pub fn starters(plan: &RoutingPlan, fallback: &str) -> Vec<String> {
 /// folds it, the conversations open and concluded, and what each seat has
 /// been shown, nudged for, or held on.
 ///
-/// Taken between waves and nowhere else. Mid-wave a conductor holds rows the
-/// host has not appended yet and a commit whose sequence it has not been
-/// told; a snapshot there would either lose those rows or duplicate them on
-/// resume, so [`Conductor::snapshot`] answers `None` until the wave settles.
+/// Carries the wave in progress too, so a snapshot is exact rather than
+/// per-wave: a host checkpoints after every committed row, and a crash
+/// replays at most the one row whose sequence had not been reported yet.
+/// The only point a snapshot cannot be taken is while the host holds a
+/// commit it has not reported -- the conductor does not know whether that
+/// row landed -- and [`Conductor::snapshot`] answers `None` there.
 ///
 /// The driver, the routing and the policy are **not** here. They are the
 /// host's to supply again on resume, exactly as they were on open: a router
@@ -150,6 +152,24 @@ pub struct ConductorState {
     waves: u64,
     /// Seats completed with their work for a spent broadcast budget.
     discharged: u64,
+    /// The wave in progress: what has been said and not yet committed, and
+    /// the steps the host has not taken. Empty between waves.
+    wave: Wave,
+}
+
+impl ConductorState {
+    /// Whether this snapshot was taken between waves, with nothing said and
+    /// nothing left for the host to do.
+    ///
+    /// A host does not need this -- [`resume_episode`] drains whatever the
+    /// wave holds either way -- but it is the difference between a restart
+    /// that lost a whole wave and one that lost a row.
+    ///
+    /// [`resume_episode`]: https://docs.rs/tinyhivemind-openhuman
+    #[must_use]
+    pub fn mid_wave_is_empty(&self) -> bool {
+        self.wave.is_idle()
+    }
 }
 
 /// The desk episode, its conversations, and the rules between them.
@@ -400,15 +420,21 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
         Ok(turns)
     }
 
-    /// Everything needed to rebuild this conductor, or `None` mid-wave.
+    /// Everything needed to rebuild this conductor, or `None` while the
+    /// host holds a commit it has not reported.
     ///
-    /// A host checkpoints where the loop already pauses: after
-    /// [`step`](Self::step) has returned `None` and before the next wave
-    /// begins. Anywhere else there are rows in flight, and the snapshot
-    /// would be a lie about what the journal holds.
+    /// A host checkpoints after every committed row. The wave in progress
+    /// travels with the snapshot, so a crash replays at most that one row:
+    /// the conductor comes back mid-wave with the same seats having spoken
+    /// and the same steps still to take.
+    ///
+    /// `None` means the host is holding a commit whose sequence it has not
+    /// reported through [`committed`](Self::committed). The conductor cannot
+    /// say whether that row reached the journal, so it will not write down a
+    /// claim either way; the caller reports the sequence and asks again.
     #[must_use]
     pub fn snapshot(&self) -> Option<ConductorState> {
-        if !self.wave.settled() {
+        if !self.wave.recordable() {
             return None;
         }
         Some(ConductorState {
@@ -427,6 +453,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
             turns: self.turns,
             waves: self.waves,
             discharged: self.discharged,
+            wave: self.wave.clone(),
         })
     }
 
@@ -434,8 +461,11 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
     /// host supplies again.
     ///
     /// The episode carries on where it paused: the same conversations are
-    /// open, the same seats are held, and a seat already nudged for its
-    /// assignment is not nudged again for it.
+    /// open, the same seats are held, a seat already nudged for its
+    /// assignment is not nudged again for it, and a wave that was in
+    /// progress resumes mid-wave. A caller drains
+    /// [`step`](Self::step) before proposing a new wave, or the restored
+    /// steps are dropped.
     ///
     /// # Errors
     ///
@@ -463,7 +493,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
             turns: snapshot.turns,
             waves: snapshot.waves,
             discharged: snapshot.discharged,
-            wave: Wave::default(),
+            wave: snapshot.wave,
         })
     }
 
diff --git a/crates/tinyhivemind-driver/src/conduct/test/parked.rs b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
index 88ec6019..873a43c1 100644
--- a/crates/tinyhivemind-driver/src/conduct/test/parked.rs
+++ b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
@@ -4,7 +4,7 @@
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
 use super::support::{
-    Journal, ask, broadcast, complete, hive, policy, seats, two_seat, wave, wave_parking,
+    Journal, ask, broadcast, complete, hive, policy, run, seats, two_seat, wave, wave_parking,
 };
 use crate::conduct::{ConductPolicy, Conductor, Event};
 use crate::driver::BroadcastRouting;
@@ -266,7 +266,7 @@ fn a_snapshot_is_taken_between_waves_and_carries_the_episode_on() {
 }
 
 #[test]
-fn a_snapshot_is_refused_while_a_commit_is_in_flight() {
+fn a_snapshot_is_refused_only_while_the_host_holds_an_unreported_commit() {
     let hive = hive(&["one", "two"]);
     let driver = CompletionDriver::new(&hive, 4).expect("driver");
     let route_policy = policy(1);
@@ -291,11 +291,26 @@ fn a_snapshot_is_refused_while_a_commit_is_in_flight() {
         &turns[0],
         [tinyhivemind::speech::ToolCall::Speak(complete("done"))],
     );
-    // A commit is queued and the host has not been told its sequence.
+    // Mid-wave, before any commit leaves: the wave travels with the
+    // snapshot, so this is recordable.
+    let held = conductor
+        .snapshot()
+        .expect("a wave in progress is still recordable");
+    assert!(
+        !held.mid_wave_is_empty(),
+        "the snapshot carries what the wave has not committed yet"
+    );
+    // The host now holds a commit whose sequence it has not reported. The
+    // conductor cannot say whether that row landed, so it will not write a
+    // claim either way.
     let step = conductor.step().expect("a step");
     assert!(matches!(step, Some(crate::Step::Commit(_))));
     assert!(
         conductor.snapshot().is_none(),
-        "mid-wave there are rows the journal does not hold yet"
+        "a commit is out there and unreported"
     );
+    // Reported, and it is recordable again.
+    let sequence = journal.append("one", "COMPLETE: done", None, None);
+    run(conductor.committed(sequence)).expect("committed");
+    assert!(conductor.snapshot().is_some());
 }
diff --git a/crates/tinyhivemind-driver/src/conduct/wave.rs b/crates/tinyhivemind-driver/src/conduct/wave.rs
index 588a418c..6d1233b2 100644
--- a/crates/tinyhivemind-driver/src/conduct/wave.rs
+++ b/crates/tinyhivemind-driver/src/conduct/wave.rs
@@ -10,6 +10,8 @@
 
 use std::collections::VecDeque;
 
+use serde::{Deserialize, Serialize};
+
 use tinyhivemind::Sequence;
 use tinyhivemind::speech::Utterance;
 use tinyhivemind_hive::CompletionEpisodeState;
@@ -22,7 +24,8 @@ use tinyhivemind::Conversation;
 
 use crate::{BoundAgent, Error, Result};
 
-#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
 enum Phase {
     /// No wave in progress.
     #[default]
@@ -40,7 +43,8 @@ enum Phase {
 }
 
 /// One wave's bookkeeping.
-#[derive(Debug, Default)]
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
 pub(super) struct Wave {
     phase: Phase,
     /// Nothing was due: every open conversation concludes without an answer.
@@ -59,16 +63,28 @@ pub(super) struct Wave {
 }
 
 impl Wave {
-    /// Nothing is in flight: no step queued, no commit queued, no commit
-    /// outstanding, and the phase machine is idle. The one point a snapshot
-    /// describes the journal truthfully.
-    pub(super) fn settled(&self) -> bool {
+    /// Whether this wave can be written down truthfully.
+    ///
+    /// Everything here is the conductor's own: a queued step or commit has
+    /// not reached the host, so a snapshot carrying it is recoverable by
+    /// re-issuing it. The one exception is `outstanding` -- a commit the
+    /// host holds and has not reported the sequence of. The conductor does
+    /// not know whether that row landed, so it is the one point a snapshot
+    /// cannot describe the journal, and the caller waits for the report.
+    /// Nothing said, nothing queued, nothing out: between waves.
+    pub(super) fn is_idle(&self) -> bool {
         matches!(self.phase, Phase::Idle)
+            && self.thread.is_empty()
+            && self.desk.is_empty()
             && self.steps.is_empty()
             && self.commits.is_empty()
             && self.outstanding.is_none()
     }
 
+    pub(super) fn recordable(&self) -> bool {
+        self.outstanding.is_none()
+    }
+
     pub(super) fn begin(&mut self, nothing_due: bool) {
         self.phase = Phase::Threads;
         self.force_conclusions = nothing_due;
diff --git a/crates/tinyhivemind-openhuman/src/episode/README.md b/crates/tinyhivemind-openhuman/src/episode/README.md
index 2847183b..9c437f4b 100644
--- a/crates/tinyhivemind-openhuman/src/episode/README.md
+++ b/crates/tinyhivemind-openhuman/src/episode/README.md
@@ -21,10 +21,17 @@ 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.
 
-`checkpoint` is handed a `ConductorState` once per wave, after the wave
-settles, and `resume_episode` carries an episode on from one: the same
-conversations open, the same seats held. A host that keeps nothing loses a
-running episode to a restart.
+`checkpoint` is handed a `ConductorState` after every committed row, and
+again at the end of each wave so a wave that only parked or nudged a seat
+is durable too. `resume_episode` carries an episode on from the newest one:
+the same conversations open, the same seats held, and a wave that was in
+progress resumed mid-wave. A host that keeps nothing loses a running
+episode to a restart.
+
+The residual window is one row: a crash between a row landing and
+`checkpoint` returning leaves the journal ahead of the snapshot, so the
+seat that wrote that row runs again. A host that cannot tolerate a
+duplicate keys its appends and drops one it has already written.
 
 `channels` names every conversation the seat is in that this episode does
 not run. Their newest rows are read through `gather_elsewhere`, bounded by
diff --git a/crates/tinyhivemind-openhuman/src/episode/mod.rs b/crates/tinyhivemind-openhuman/src/episode/mod.rs
index 1111918f..753046f9 100644
--- a/crates/tinyhivemind-openhuman/src/episode/mod.rs
+++ b/crates/tinyhivemind-openhuman/src/episode/mod.rs
@@ -80,13 +80,19 @@ pub trait Journal: Send + Sync {
         Vec::new()
     }
 
-    /// The episode paused where it can be resumed: every row it committed
-    /// is in the journal and nothing is in flight. The default keeps
+    /// The episode is exactly where this snapshot says. The default keeps
     /// nothing, and such a host loses a running episode to a restart.
     ///
-    /// Called once per wave, after the wave settles. A host stores the
-    /// snapshot beside its rows and hands it to
-    /// [`resume_episode`] on boot.
+    /// Called after **every committed row**, not once per wave: the row is
+    /// in the journal and the conductor has folded it, so the two agree.
+    /// A host stores the snapshot beside its rows -- ideally in the same
+    /// journal, so the ordering is the journal's own -- and hands the
+    /// newest one to [`resume_episode`] on boot.
+    ///
+    /// A crash between a row landing and this returning replays that one
+    /// row: the resumed conductor has not folded it, so the seat that wrote
+    /// it runs again. A host that cannot tolerate a duplicate row keys its
+    /// appends and drops one it has already written.
     ///
     /// # Errors
     ///
@@ -227,6 +233,10 @@ where
     J: Journal,
     R: SeatRunner,
 {
+    // A snapshot taken mid-wave comes back mid-wave: drain what it restored
+    // before proposing anything, because `begin_wave` resets the phase and
+    // would drop those steps on the floor.
+    settle_wave(journal, &mut conductor).await?;
     loop {
         if conductor.finished() {
             break;
@@ -271,11 +281,10 @@ where
                 }
             }
         }
-        while let Some(step) = conductor.step()? {
-            settle(journal, &mut conductor, step).await?;
-        }
-        // The wave settled: every row it produced is in the journal and
-        // nothing is in flight, which is the one point a snapshot is true.
+        settle_wave(journal, &mut conductor).await?;
+        // And once the wave is over, whether or not it committed anything:
+        // a wave that only parked a seat or nudged one moved state no row
+        // records, and a restart would otherwise lose it.
         if let Some(snapshot) = conductor.snapshot() {
             journal.checkpoint(&snapshot)?;
         }
@@ -289,6 +298,26 @@ where
     })
 }
 
+/// Take every step the wave has left, checkpointing after each committed
+/// row so a crash replays at most that one row.
+async fn settle_wave<A: BoundAgent, J: Journal>(
+    journal: &J,
+    conductor: &mut Conductor<'_, A>,
+) -> Result<()> {
+    while let Some(step) = conductor.step()? {
+        let committed = matches!(step, Step::Commit(_));
+        settle(journal, conductor, step).await?;
+        // After a commit the row is in the journal and the conductor has
+        // folded it, so the snapshot and the journal agree. A note or an
+        // event moves nothing a resume would double-count, so neither is
+        // worth a write.
+        if committed && let Some(snapshot) = conductor.snapshot() {
+            journal.checkpoint(&snapshot)?;
+        }
+    }
+    Ok(())
+}
+
 /// One step taken: a commit appended and its sequence reported, a note
 /// appended, an event shown.
 async fn settle<A: BoundAgent, J: Journal>(
diff --git a/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs b/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs
index 7eec4cdb..2a3b2e76 100644
--- a/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs
+++ b/crates/tinyhivemind-openhuman/src/episode/test/watermark.rs
@@ -11,7 +11,9 @@ use tinyhivemind_driver::{BroadcastRouting, CompletionDriver, ConductPolicy, Doo
 use super::super::{resume_episode, run_episode};
 use serde_json::json;
 
-use super::support::{GrowingLog, ScriptRunner, TestJournal, complete, door, hive, policy, run};
+use super::support::{
+    GrowingLog, ScriptRunner, TestJournal, ask, complete, door, hive, policy, run,
+};
 use crate::journal::MemoryLog;
 
 #[test]
@@ -141,7 +143,7 @@ fn an_episode_is_checkpointed_every_wave_and_carries_on_from_one() {
         .unwrap()
         .last()
         .cloned()
-        .expect("a wave settled before the episode stopped");
+        .expect("a row was committed before the episode stopped");
     assert_eq!(snapshot.chat, "engineering");
 
     // A fresh process: the same journal, a new runner, resumed from the
@@ -167,3 +169,47 @@ fn an_episode_is_checkpointed_every_wave_and_carries_on_from_one() {
         after.prompts()
     );
 }
+
+#[test]
+fn a_crash_after_a_row_lands_replays_that_row_and_no_more() {
+    let hive = hive(&["one", "two"]);
+    let driver = CompletionDriver::new(&hive, 4).expect("driver");
+    let route_policy = policy();
+    let routing = || BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    let journal = TestJournal::new();
+    // One asks two: the ask is a committed row, and the wave that carries
+    // it has more to do afterwards.
+    let runner = ScriptRunner::new(
+        &["one", "two"],
+        &[("one", vec![vec![ask("two", "which port?", None)]])],
+    );
+    let entrance = door(&journal, &["one", "two"], &["one"]);
+    // Stop the episode the moment the ask has been committed, by refusing
+    // to release anybody once nothing is due.
+    let _ = run(run_episode(
+        &journal,
+        &runner,
+        &driver,
+        routing(),
+        ConductPolicy::default(),
+        entrance,
+    ));
+    let checkpoints = journal.checkpoints.lock().unwrap().clone();
+    assert!(
+        checkpoints.len() > 1,
+        "a checkpoint per committed row, not one per wave: {}",
+        checkpoints.len()
+    );
+    // The first checkpoint is taken mid-wave: the wave it carries is not
+    // idle, which is what bounds a crash to one row.
+    assert!(
+        checkpoints.iter().any(|state| !state.mid_wave_is_empty()),
+        "at least one checkpoint was taken with the wave still in progress"
+    );
+}

From 116a86239d87016e326d0e703f89993be652d3db Mon Sep 17 00:00:00 2001
From: sanil-23 <sanil@alphahuman.xyz>
Date: Wed, 23 Sep 2026 12:21:46 +0530
Subject: [PATCH 3/4] Keep a held seat from ending the episode, and pin the
 snapshot's wire form

`finished` was quiescence plus no open conversation, which let an episode
end with a seat still held on the host. A seat that broadcasts and parks
in one turn can have its assignment discharged by that broadcast, so the
desk goes quiescent while the host is still holding it; the loop broke
out, and the approval the host queued had nowhere to return to when the
operator answered it. A held seat now keeps the episode open, and
releasing the last one is allowed to be what finishes it -- the loop asks
whether the episode is over before asking for turns, because an empty
wave with nothing parked reads as a stall.

`ConductorState` is written down and read back across restarts and across
upgrades, so its field names are the wire form, and only a same-process
round trip covered them. A rename would have left every test green and
every stored snapshot undecodable, losing exactly the episodes this work
exists to save. Its keys are pinned now, as every other payload here is.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../tinyhivemind-driver/src/conduct/README.md |  5 +-
 crates/tinyhivemind-driver/src/conduct/mod.rs |  7 +-
 .../src/conduct/test/parked.rs                | 35 +++++++++
 .../src/conduct/test/wire.rs                  | 71 +++++++++++++++++++
 .../tinyhivemind-openhuman/src/episode/mod.rs |  7 ++
 5 files changed, 123 insertions(+), 2 deletions(-)

diff --git a/crates/tinyhivemind-driver/src/conduct/README.md b/crates/tinyhivemind-driver/src/conduct/README.md
index 419c950d..43f764b8 100644
--- a/crates/tinyhivemind-driver/src/conduct/README.md
+++ b/crates/tinyhivemind-driver/src/conduct/README.md
@@ -30,7 +30,10 @@ The rules, each with the decision it comes from:
   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.
+  it parked. `Event::Parked` and `Event::Resumed` mark both. A held seat
+  also keeps `finished` false even where its own work closed some other way,
+  because ending the episode there would strand whatever the host queued to
+  hold it: the operator answers an approval with no loop to return to.
 - **Checkpointing**: `snapshot` carries the wave in progress as well as the
   episode, so a host checkpoints after **every committed row** rather than
   once per wave, and a crash replays at most the one row whose sequence had
diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs
index a90dd3bf..83d8c9c0 100644
--- a/crates/tinyhivemind-driver/src/conduct/mod.rs
+++ b/crates/tinyhivemind-driver/src/conduct/mod.rs
@@ -276,7 +276,12 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
     /// Over: the desk is quiescent and no conversation is open.
     #[must_use]
     pub fn finished(&self) -> bool {
-        self.state.quiescent() && self.children.is_empty()
+        // A parked seat is not a finished one, even where its work closed
+        // some other way -- a broadcast it made in the same turn spending
+        // the budget, say. Ending the episode there would strand whatever
+        // the host queued to hold it: the operator answers an approval that
+        // no longer has a loop to return to.
+        self.state.quiescent() && self.children.is_empty() && self.parked.is_empty()
     }
 
     /// The desk episode's state.
diff --git a/crates/tinyhivemind-driver/src/conduct/test/parked.rs b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
index 873a43c1..ad3bdf18 100644
--- a/crates/tinyhivemind-driver/src/conduct/test/parked.rs
+++ b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
@@ -314,3 +314,38 @@ fn a_snapshot_is_refused_only_while_the_host_holds_an_unreported_commit() {
     run(conductor.committed(sequence)).expect("committed");
     assert!(conductor.snapshot().is_some());
 }
+
+#[test]
+fn a_held_seat_keeps_the_episode_open_even_where_its_work_closed() {
+    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 completes and parks in the same turn: its work is closed, so the
+    // desk is quiescent, but the host is still holding it.
+    wave_parking(
+        &mut conductor,
+        &journal,
+        &[("one", vec![complete("done")])],
+        &["one"],
+    )
+    .expect("wave");
+    assert_eq!(conductor.parked(), vec!["one".to_owned()]);
+    assert!(
+        !conductor.finished(),
+        "a held seat is not a finished one: the operator's answer needs a \
+         loop to come back to"
+    );
+    // Released, there is nothing left to hold and the episode is over.
+    conductor.resume_seat("one");
+    assert!(conductor.parked().is_empty());
+    assert!(conductor.finished());
+}
diff --git a/crates/tinyhivemind-driver/src/conduct/test/wire.rs b/crates/tinyhivemind-driver/src/conduct/test/wire.rs
index 25fb0d33..b63293a0 100644
--- a/crates/tinyhivemind-driver/src/conduct/test/wire.rs
+++ b/crates/tinyhivemind-driver/src/conduct/test/wire.rs
@@ -246,3 +246,74 @@ fn every_event_and_refusal_survives_the_wire() {
         json!({"kind": "not_yet_shown"})
     );
 }
+
+#[test]
+fn a_conductor_snapshot_names_every_field_a_restart_reads_back() {
+    let hive = super::support::hive(&["one", "two"]);
+    let driver = crate::CompletionDriver::new(&hive, 4).expect("driver");
+    let route_policy = super::support::policy(1);
+    let routing = crate::driver::BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    let journal = super::support::Journal::default();
+    let mut conductor =
+        super::support::two_seat(&driver, routing, crate::ConductPolicy::default(), &journal);
+    // A conversation open and a seat held: the two pieces of state a
+    // restart most needs, so both are on the wire.
+    super::support::wave(
+        &mut conductor,
+        &journal,
+        &[("one", vec![super::support::ask("two", "which port?")])],
+    )
+    .expect("wave");
+    let snapshot = conductor.snapshot().expect("recordable");
+    let wire = serde_json::to_value(&snapshot).expect("serializes");
+    let object = wire.as_object().expect("an object");
+
+    // The spelling a stored snapshot is read back by. Renaming any of these
+    // orphans every snapshot written by the build before it, which is the
+    // failure the whole checkpoint exists to prevent.
+    let mut keys: Vec<&str> = object.keys().map(String::as_str).collect();
+    keys.sort_unstable();
+    assert_eq!(
+        keys,
+        [
+            "chat",
+            "children",
+            "concluded",
+            "desk_name",
+            "desk_nudged",
+            "discharged",
+            "parked",
+            "shown",
+            "state",
+            "turns",
+            "wave",
+            "waves",
+        ]
+    );
+    assert_eq!(object["chat"], json!("engineering"));
+
+    // A conversation on the wire is its root paired with its record, and
+    // the record spells out both seats and the nudge it may still owe.
+    let child = &wire["children"][0][1];
+    for field in [
+        "root", "asker", "askee", "state", "turns", "nudged", "turned",
+    ] {
+        assert!(
+            child.get(field).is_some(),
+            "a conversation must carry `{field}`: {child}"
+        );
+    }
+    assert_eq!(child["asker"], json!("one"));
+    assert_eq!(child["askee"], json!("two"));
+
+    // And it decodes back to the same thing.
+    let back: crate::ConductorState = serde_json::from_value(wire).expect("deserializes");
+    assert_eq!(back.chat, snapshot.chat);
+    assert_eq!(back.mid_wave_is_empty(), snapshot.mid_wave_is_empty());
+}
diff --git a/crates/tinyhivemind-openhuman/src/episode/mod.rs b/crates/tinyhivemind-openhuman/src/episode/mod.rs
index 753046f9..3f3fd156 100644
--- a/crates/tinyhivemind-openhuman/src/episode/mod.rs
+++ b/crates/tinyhivemind-openhuman/src/episode/mod.rs
@@ -384,6 +384,13 @@ async fn wait_for_release<A: BoundAgent, J: Journal>(
     for seat in &released {
         conductor.resume_seat(seat);
     }
+    // Releasing the last held seat can be the thing that finishes the
+    // episode -- its work may have closed while it was held. Ask again only
+    // if there is still an episode to ask about; `turns` would read an
+    // empty wave with nothing parked as a stall.
+    if conductor.finished() {
+        return Ok(Vec::new());
+    }
     Ok(conductor.turns()?)
 }
 

From 808b3a4551c5a2aae79336382c3aab3ddef26065 Mon Sep 17 00:00:00 2001
From: sanil-23 <sanil@alphahuman.xyz>
Date: Wed, 23 Sep 2026 12:35:37 +0530
Subject: [PATCH 4/4] Validate a snapshot on the way in rather than trusting it

`resume` put the desk episode through `driver.resume` and then copied
everything else in raw. A snapshot is the host's file, not the
conductor's memory, so a stale or crafted one could restore a
conversation naming another desk or an unknown seat, a cursor pointing
past the conversations it indexes -- which panics the first time that
seat is briefed -- or a seat held that this desk does not seat, which
nothing could ever release. Each failed later, where the cause is
unrecoverable, or not at all.

Every conversation now goes through the same `driver.resume` the desk
episode does, a conversation filed under a root it does not call its own
is refused, and so are an out-of-range cursor and a held seat that is not
seated. `Error::InconsistentSnapshot` says which.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 crates/tinyhivemind-driver/src/conduct/mod.rs | 57 +++++++++++++-
 .../src/conduct/test/parked.rs                | 78 +++++++++++++++++++
 .../src/conduct/test/wire.rs                  | 12 ++-
 crates/tinyhivemind-driver/src/error/mod.rs   | 13 ++++
 .../src/episode/README.md                     |  6 +-
 5 files changed, 157 insertions(+), 9 deletions(-)

diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs
index 83d8c9c0..1b841bee 100644
--- a/crates/tinyhivemind-driver/src/conduct/mod.rs
+++ b/crates/tinyhivemind-driver/src/conduct/mod.rs
@@ -474,8 +474,12 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
     ///
     /// # Errors
     ///
-    /// The driver refusing the episode -- a state naming a seat or a desk
-    /// this hive does not have.
+    /// The driver refusing the episode or any of its conversations -- a
+    /// state naming a seat or a desk this hive does not have -- and
+    /// [`Error::InconsistentSnapshot`] for a snapshot that disagrees with
+    /// itself: a conversation filed under the wrong root, a cursor past the
+    /// conversations it points into, or a seat held that this desk does not
+    /// seat.
     pub fn resume(
         driver: &'a CompletionDriver<'a, A>,
         routing: BroadcastRouting<'a>,
@@ -483,6 +487,53 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
         snapshot: ConductorState,
     ) -> Result<Self> {
         let state = driver.resume(snapshot.state)?;
+        // A snapshot is the host's file, not the conductor's memory: every
+        // part of it is validated here rather than trusted, because the
+        // alternative is an episode that resumes and then misbehaves waves
+        // later with nothing left to say why.
+        let mut children = BTreeMap::new();
+        for (root, mut child) in snapshot.children {
+            if root != child.root {
+                return Err(Error::InconsistentSnapshot {
+                    reason: format!(
+                        "a conversation filed under {} calls itself {}",
+                        root.0, child.root.0
+                    ),
+                });
+            }
+            // The same validation the desk episode gets: the conversation
+            // names this desk, and its two seats are seats of this hive.
+            child.state = driver.resume(child.state)?;
+            children.insert(root, child);
+        }
+        // A cursor into the concluded list, for a seat that has been shown
+        // some of them. Past the end it would panic the first time that
+        // seat is briefed.
+        for (seat, cursor) in &snapshot.shown {
+            if *cursor > snapshot.concluded.len() {
+                return Err(Error::InconsistentSnapshot {
+                    reason: format!(
+                        "@{seat} has been shown {cursor} conversations of {}",
+                        snapshot.concluded.len()
+                    ),
+                });
+            }
+        }
+        // A seat held on the host that this desk does not seat would be held
+        // for ever: it is never proposed, and nothing can release it into a
+        // wave.
+        for seat in snapshot.parked.keys() {
+            if !state
+                .episode()
+                .participants
+                .iter()
+                .any(|participant| &participant.agent_id == seat)
+            {
+                return Err(Error::InconsistentSnapshot {
+                    reason: format!("@{seat} is held but is not seated at this desk"),
+                });
+            }
+        }
         Ok(Self {
             driver,
             routing,
@@ -490,7 +541,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
             desk_name: snapshot.desk_name,
             policy,
             state,
-            children: snapshot.children.into_iter().collect(),
+            children,
             concluded: snapshot.concluded,
             shown: snapshot.shown,
             desk_nudged: snapshot.desk_nudged,
diff --git a/crates/tinyhivemind-driver/src/conduct/test/parked.rs b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
index ad3bdf18..91e90a3a 100644
--- a/crates/tinyhivemind-driver/src/conduct/test/parked.rs
+++ b/crates/tinyhivemind-driver/src/conduct/test/parked.rs
@@ -8,7 +8,9 @@ use super::support::{
 };
 use crate::conduct::{ConductPolicy, Conductor, Event};
 use crate::driver::BroadcastRouting;
+use crate::test_support::Seat;
 use crate::{CompletionDriver, Error};
+use serde_json::{Value, json};
 use tinyhivemind::Sequence;
 
 #[test]
@@ -349,3 +351,79 @@ fn a_held_seat_keeps_the_episode_open_even_where_its_work_closed() {
     assert!(conductor.parked().is_empty());
     assert!(conductor.finished());
 }
+
+/// A snapshot with one conversation open and one seat held, as a value a
+/// test can bend before handing it back.
+fn snapshot_with_a_conversation(journal: &Journal, conductor: &mut Conductor<'_, Seat>) -> Value {
+    wave(
+        conductor,
+        journal,
+        &[("one", vec![ask("two", "which port?")])],
+    )
+    .expect("wave");
+    serde_json::to_value(conductor.snapshot().expect("recordable")).expect("serializes")
+}
+
+#[test]
+fn a_snapshot_that_disagrees_with_itself_is_refused_rather_than_resumed() {
+    let hive = hive(&["one", "two"]);
+    let driver = CompletionDriver::new(&hive, 4).expect("driver");
+    let route_policy = policy(1);
+    let routing = || BroadcastRouting {
+        primary: None,
+        reasoning: None,
+        policy: &route_policy,
+        roster_version: 1,
+        thread_context: &[],
+    };
+    let journal = Journal::default();
+    let mut conductor = two_seat(&driver, routing(), ConductPolicy::default(), &journal);
+    let good = snapshot_with_a_conversation(&journal, &mut conductor);
+    // The unbent snapshot resumes, so each refusal below is about the bend.
+    let restored: crate::ConductorState =
+        serde_json::from_value(good.clone()).expect("deserializes");
+    assert!(Conductor::resume(&driver, routing(), ConductPolicy::default(), restored).is_ok());
+
+    // A conversation filed under a root it does not call its own.
+    let mut bent = good.clone();
+    bent["children"][0][0] = json!(99);
+    let restored: crate::ConductorState = serde_json::from_value(bent).expect("deserializes");
+    let refused = Conductor::resume(&driver, routing(), ConductPolicy::default(), restored);
+    assert!(
+        matches!(&refused, Err(Error::InconsistentSnapshot { reason }) if reason.contains("99")),
+        "{refused:?}"
+    );
+
+    // A cursor past the conversations it points into: shown_conversations
+    // would index off the end of the list the first time that seat spoke.
+    let mut bent = good.clone();
+    bent["shown"] = json!({ "one": 7 });
+    let restored: crate::ConductorState = serde_json::from_value(bent).expect("deserializes");
+    let refused = Conductor::resume(&driver, routing(), ConductPolicy::default(), restored);
+    assert!(
+        matches!(&refused, Err(Error::InconsistentSnapshot { reason }) if reason.contains("@one")),
+        "{refused:?}"
+    );
+
+    // A seat held that this desk does not seat: nothing could ever release
+    // it into a wave.
+    let mut bent = good.clone();
+    bent["parked"] = json!({ "nobody": null });
+    let restored: crate::ConductorState = serde_json::from_value(bent).expect("deserializes");
+    let refused = Conductor::resume(&driver, routing(), ConductPolicy::default(), restored);
+    assert!(
+        matches!(&refused, Err(Error::InconsistentSnapshot { reason }) if reason.contains("nobody")),
+        "{refused:?}"
+    );
+
+    // A conversation whose own state names another desk is refused by the
+    // driver, exactly as the desk episode's state would be.
+    let mut bent = good;
+    bent["children"][0][1]["state"]["episode"]["conversation"]["desk_id"] = json!("marketing");
+    let restored: crate::ConductorState = serde_json::from_value(bent).expect("deserializes");
+    let refused = Conductor::resume(&driver, routing(), ConductPolicy::default(), restored);
+    assert!(
+        matches!(&refused, Err(Error::OutOfHiveEpisode { .. })),
+        "{refused:?}"
+    );
+}
diff --git a/crates/tinyhivemind-driver/src/conduct/test/wire.rs b/crates/tinyhivemind-driver/src/conduct/test/wire.rs
index b63293a0..dced3b42 100644
--- a/crates/tinyhivemind-driver/src/conduct/test/wire.rs
+++ b/crates/tinyhivemind-driver/src/conduct/test/wire.rs
@@ -298,9 +298,15 @@ fn a_conductor_snapshot_names_every_field_a_restart_reads_back() {
     );
     assert_eq!(object["chat"], json!("engineering"));
 
-    // A conversation on the wire is its root paired with its record, and
-    // the record spells out both seats and the nudge it may still owe.
-    let child = &wire["children"][0][1];
+    // A conversation on the wire is its root paired with its record: the
+    // key is the root a host files it under, and `resume` refuses a pair
+    // whose two halves disagree.
+    let pair = &wire["children"][0];
+    let child = &pair[1];
+    assert_eq!(
+        pair[0], child["root"],
+        "the key a conversation is filed under is its own root"
+    );
     for field in [
         "root", "asker", "askee", "state", "turns", "nudged", "turned",
     ] {
diff --git a/crates/tinyhivemind-driver/src/error/mod.rs b/crates/tinyhivemind-driver/src/error/mod.rs
index a889712e..57dad79a 100644
--- a/crates/tinyhivemind-driver/src/error/mod.rs
+++ b/crates/tinyhivemind-driver/src/error/mod.rs
@@ -156,6 +156,19 @@ pub enum Error {
         /// Number of committed receipts represented by the state.
         receipt_count: usize,
     },
+    /// A restored conversation disagrees with the snapshot that carries it:
+    /// its key is not its root, its seats are not the two it names, or a
+    /// cursor points past what it points into.
+    ///
+    /// A snapshot is host-stored data, so it is validated on the way in
+    /// rather than trusted: the alternative is an episode that resumes and
+    /// then panics or misbehaves several waves later, where the cause is
+    /// unrecoverable.
+    #[error("restored episode state is inconsistent: {reason}")]
+    InconsistentSnapshot {
+        /// What disagreed.
+        reason: String,
+    },
     /// A serialized freshness floor disagrees with its episode and receipts.
     #[error("driver freshness floor {stored} does not match derived floor {derived}")]
     InvalidFreshnessFloor {
diff --git a/crates/tinyhivemind-openhuman/src/episode/README.md b/crates/tinyhivemind-openhuman/src/episode/README.md
index 9c437f4b..3117220f 100644
--- a/crates/tinyhivemind-openhuman/src/episode/README.md
+++ b/crates/tinyhivemind-openhuman/src/episode/README.md
@@ -9,11 +9,11 @@ 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 five optional hooks -- `event` to show what the episode did, `compose`
+and six 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, `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.
+conversations, `released` to say which parked seats the host has settled,
+and `checkpoint` to keep the snapshot a restart resumes from. `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