Run the episode on the host's own agents: the hosted runner

crates/tinyhivemind-driver/src/conduct/README.md

@@ -9,8 +9,8 @@ that no single fold can hold.
 | `mod.rs` | `Conductor`, `ConductPolicy`, `Door`, `starters`; opening the desk, beginning a wave, proposing turns, opening a turn with its brief, recording what it called |
 | `wave.rs` | After a wave: the phase machine that hands the host one `Step` at a time -- commits in conversations, silent askees, commits on the desk with their consequences, conclusions, the turn wall |
 | `child.rs` | A conversation: its root, its two seats, its own driver state, its turns, its nudge; and one that concluded |
-| `steps.rs` | `Turn`, `Note`, `Commit`, `Event`, `Refusal`, `Step` |
-| `test.rs` | Every rule, driven by a host that is only a journal |
+| `steps.rs` | `Turn`, `Note`, `Commit`, `Event`, `Refusal`, `Step`: the wire forms a host journals and streams |
+| `test/` | Every rule, driven by a host that is only a journal; the exact wire forms; the links from a row to its conversation and from an event to its row |
 
 The rules, each with the decision it comes from:
 
@@ -33,3 +33,16 @@ The conductor appends nothing. It hands the host a `Note` to append, a
 `Commit` to append and report the sequence of, or an `Event` to log, and
 takes the sequence back through `committed`. The host owns the journal,
 the rendering of a row, the prompt, and running the turn.
+
+What a host reads back to draw the desk:
+
+- **A conversation, whole.** The ask row's sequence is the conversation's
+  root. Every other row of it carries that root as `Commit::conversation`:
+  what was said inside it, desk work a seat lifted out of it, and the row
+  that concluded it to the asker. `Event::Asked` and `Event::Concluded`
+  mark when it opened and closed.
+- **An event's row.** `Broadcast`, `Unplaced`, `CompletedByBroadcast`,
+  `Refused`, `Discharged` and `Concluded` carry `at`, the sequence the host
+  gave the row they are about; `Handoff` carries the broadcast row it came
+  from as `origin`. A refused row is already on the journal, and its event
+  is what marks it refused.

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

@@ -441,8 +441,11 @@ impl<'a, A: BoundAgent> Conductor<'a, A> {
                 continue;
             };
             match (turn.thread(), &utterance) {
-                (None, _) | (Some(_), Utterance::Broadcast { .. } | Utterance::Ask { .. }) => {
-                    self.wave.desk.push((turn.seat.clone(), utterance));
+                (None, _) => self.wave.desk.push((turn.seat.clone(), utterance, None)),
+                (Some(root), Utterance::Broadcast { .. } | Utterance::Ask { .. }) => {
+                    self.wave
+                        .desk
+                        .push((turn.seat.clone(), utterance, Some(root)));
                 }
                 (Some(root), _) => self.wave.thread.push((root, turn.seat.clone(), utterance)),
             }

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

@@ -1,13 +1,20 @@
 //! What passes between the conductor and its host: a turn to run, and the
 //! steps the host takes on the conductor's behalf after a wave.
+//!
+//! Every type here is a wire form. A host journals commits and events and
+//! streams them to whatever renders the desk, so the serde representation is
+//! pinned by a unit test: internally tagged, `snake_case`, and every field a
+//! host can act on present by name.
 
+use serde::{Deserialize, Serialize};
 use tinyhivemind::Sequence;
 use tinyhivemind::speech::Utterance;
 
 use crate::driver::Channel;
 
 /// One turn the host runs this wave.
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
 pub struct Turn {
     /// The seat.
     pub seat: String,
@@ -31,7 +38,8 @@ impl Turn {
 
 /// A row the desk says to a seat: the host appends it, attributed to the
 /// desk, and nothing is committed back. The wording is the episode's.
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
 pub struct Note {
     /// What the desk says.
     pub body: String,
@@ -46,7 +54,8 @@ pub struct Note {
 ///
 /// The host renders the utterance as its own desk row and calls
 /// [`Conductor::committed`](super::Conductor::committed) with the sequence.
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
 pub struct Commit {
     /// The seat the row is attributed to.
     pub author: String,
@@ -56,14 +65,29 @@ pub struct Commit {
     pub thread: Option<Sequence>,
     /// On the open desk, the one seat it reaches; `None` reaches every seat.
     pub only_for: Option<String>,
+    /// The conversation this row belongs to, by the ask row it is rooted at.
+    ///
+    /// Set for a row said inside a conversation, for desk work a seat lifted
+    /// out of one -- a broadcast or an ask made while talking, which lands on
+    /// the desk with no `thread` -- and for the row that concludes one to its
+    /// asker. `None` for everything said on the desk itself, including the
+    /// ask that opens a conversation: that row's own sequence is the root.
+    /// A host shows one agent-to-agent exchange whole by taking the ask row
+    /// and every row whose `conversation` is its sequence, wherever they
+    /// landed.
+    pub conversation: Option<Sequence>,
+    /// What the conductor does with the row once it has its sequence. Opaque
+    /// to a host: carried so a commit survives the wire whole.
+    #[serde(rename = "purpose")]
     pub(super) kind: Kind,
 }
 
 /// What the conductor does with a commit once it has its sequence.
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
 pub(super) enum Kind {
     /// A seat spoke in a conversation.
-    Thread(Sequence),
+    Thread { root: Sequence },
     /// A seat spoke on the desk; a broadcast is routed.
     Desk,
     /// A conversation concluded: its outcome, cross-posted to the asker.
@@ -73,7 +97,8 @@ pub(super) enum Kind {
 }
 
 /// Why a seat's row was refused, in the terms the desk tells it.
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
 pub enum Refusal {
     /// It may not complete: the seats it asked have not answered.
     AwaitingReply {
@@ -89,9 +114,14 @@ pub enum Refusal {
     NotYetShown,
 }
 
-/// Something the episode did that a host may want to log. Nothing here needs
-/// acting on; every consequence is already a [`Note`] or a [`Commit`].
-#[derive(Clone, Debug, Eq, PartialEq)]
+/// Something the episode did that a host may want to show. Nothing here
+/// needs acting on; every consequence is already a [`Note`] or a [`Commit`].
+///
+/// An event about a row names it by `at`, the sequence the host gave that
+/// row, so a host attaches the event to the row it drew rather than
+/// inferring it from order.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
 pub enum Event {
     /// A seat was told once that nothing will wake it.
     Nudged {
@@ -106,16 +136,22 @@ pub enum Event {
         seat: String,
         /// Who took it.
         to: Vec<String>,
+        /// The broadcast row.
+        at: Sequence,
     },
     /// A broadcast fit no seat; the author keeps the work.
     Unplaced {
         /// The author.
         seat: String,
+        /// The broadcast row.
+        at: Sequence,
     },
     /// A broadcast closed its author's own assignment.
     CompletedByBroadcast {
         /// The author.
         seat: String,
+        /// The broadcast row.
+        at: Sequence,
     },
     /// An ask opened a conversation.
     Asked {
@@ -132,20 +168,27 @@ pub enum Event {
         to: String,
         /// The author of the broadcast it came from.
         from: String,
+        /// The broadcast row it came from.
+        origin: Sequence,
     },
-    /// A row was refused, and the seat told why.
+    /// A row was refused, and the seat told why. The row is already on the
+    /// host's journal; this is what marks it refused.
     Refused {
         /// The seat.
         seat: String,
         /// The thread, or `None` on the desk.
         thread: Option<Sequence>,
         /// Why.
         why: Refusal,
+        /// The refused row.
+        at: Sequence,
     },
     /// A seat spent its broadcast budget and was completed with the work.
     Discharged {
         /// The seat.
         seat: String,
+        /// The broadcast row that was over budget.
+        at: Sequence,
     },
     /// A conversation concluded.
     Concluded {
@@ -157,16 +200,21 @@ pub enum Event {
         askee: String,
         /// Without an answer: nothing was due, or it ran out of turns.
         forced: bool,
+        /// The row that carried the outcome to the asker.
+        at: Sequence,
     },
 }
 
 /// One step the host takes after a wave, in order.
-#[derive(Clone, Debug, Eq, PartialEq)]
+///
+/// On the wire, tagged by `step`, with the step's own fields beside the tag.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "step", rename_all = "snake_case")]
 pub enum Step {
     /// Append this, attributed to the desk.
     Note(Note),
     /// Append this and report its sequence.
     Commit(Commit),
-    /// Log this, or don't.
+    /// Show this, or don't.
     Event(Event),
 }

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

@@ -59,7 +59,7 @@ fn an_ask_opens_a_conversation_that_runs_first_and_concludes_to_the_asker() {
     ));
     assert!(matches!(
         answered.events.as_slice(),
-        [Event::Concluded { root: at, asker, askee, forced: false }]
+        [Event::Concluded { root: at, asker, askee, forced: false, .. }]
             if *at == root && asker == "one" && askee == "two"
     ));
     assert_eq!(conductor.conversations(), 1);
@@ -122,7 +122,7 @@ fn a_completion_while_a_conversation_is_open_is_refused_and_explained() {
     .expect("wave");
     assert!(seen.events.iter().any(|event| matches!(
         event,
-        Event::Refused { seat, thread: None, why: Refusal::AwaitingReply { waiting_on } }
+        Event::Refused { seat, thread: None, why: Refusal::AwaitingReply { waiting_on }, .. }
             if seat == "one" && waiting_on == &["two".to_owned()]
     )));
     assert!(
@@ -370,7 +370,7 @@ fn a_broadcast_or_ask_inside_a_conversation_is_desk_work_and_a_dm_is_dropped() {
     assert!(
         seen.events
             .iter()
-            .any(|event| matches!(event, Event::Broadcast { seat, to } if seat == "two" && !to.is_empty())),
+            .any(|event| matches!(event, Event::Broadcast { seat, to, .. } if seat == "two" && !to.is_empty())),
         "{:?}",
         seen.events
     );
@@ -423,3 +423,67 @@ fn nothing_due_concludes_every_open_conversation_without_an_answer() {
     );
     assert_eq!(conductor.conversations(), 1);
 }
+
+#[test]
+fn a_conclusion_the_fold_refuses_leaves_the_conversation_to_conclude_later() {
+    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 asked = wave(&mut conductor, &journal, &[("one", vec![ask("two", "?")])]).expect("wave");
+    let root = asked.commits[0].0;
+
+    // The askee answers, and the host reports the conclusion's row at a
+    // sequence the episode already holds: the fold refuses it.
+    conductor.begin_wave();
+    let turns = conductor.turns().expect("turns");
+    for turn in &turns {
+        conductor.open_turn(turn, journal.latest(), Vec::new(), |root| {
+            journal.thread(root)
+        });
+        if turn.seat == "two" {
+            conductor.record(turn, vec![ToolCall::Speak(complete("port 8080"))]);
+        }
+    }
+    let mut refused = false;
+    while let Some(step) = conductor.step().expect("steps") {
+        if let Step::Commit(commit) = step {
+            let sequence = if matches!(commit.utterance, Utterance::Dm { .. }) {
+                root
+            } else {
+                journal.append(
+                    &commit.author,
+                    "row",
+                    commit.thread,
+                    commit.only_for.clone(),
+                )
+            };
+            if run(conductor.committed(sequence)).is_err() {
+                refused = true;
+            }
+        }
+    }
+    assert!(refused, "a reused sequence is refused by the fold");
+    assert_eq!(conductor.conversations(), 0, "nothing concluded");
+    assert!(!conductor.finished(), "the conversation is still open");
+
+    // The next wave concludes it, at a row the host gives properly.
+    let later = wave(&mut conductor, &journal, &[]).expect("wave");
+    assert!(
+        later
+            .events
+            .iter()
+            .any(|event| matches!(event, Event::Concluded { root: at, .. } if *at == root)),
+        "{:?}",
+        later.events
+    );
+    assert_eq!(conductor.conversations(), 1);
+}

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

@@ -65,7 +65,7 @@ fn an_unplaced_broadcast_leaves_the_work_with_the_author_and_says_so() {
     .expect("wave");
     assert!(matches!(
         seen.events.as_slice(),
-        [Event::Unplaced { seat }] if seat == "one"
+        [Event::Unplaced { seat, .. }] if seat == "one"
     ));
     assert!(journal.private_to("one")[0].contains("nobody on this desk can take that"));
     assert!(!conductor.finished(), "the author keeps the work");
@@ -104,7 +104,7 @@ fn a_placed_broadcast_completes_its_author_and_a_busy_recipient_gets_it_as_a_han
     assert!(
         seen.events.iter().any(|event| matches!(
             event,
-            Event::CompletedByBroadcast { seat } if seat == "one"
+            Event::CompletedByBroadcast { seat, .. } if seat == "one"
         )),
         "{:?}",
         seen.events
@@ -119,7 +119,7 @@ fn a_placed_broadcast_completes_its_author_and_a_busy_recipient_gets_it_as_a_han
     assert!(
         handed.events.iter().any(|event| matches!(
             event,
-            Event::Handoff { to, from } if to == "two" && from == "one"
+            Event::Handoff { to, from, .. } if to == "two" && from == "one"
         )),
         "{:?}",
         handed.events
@@ -174,7 +174,7 @@ fn a_spent_broadcast_budget_completes_the_seat_with_the_work() {
     assert!(
         seen.events
             .iter()
-            .any(|event| matches!(event, Event::Discharged { seat } if seat == "one")),
+            .any(|event| matches!(event, Event::Discharged { seat, .. } if seat == "one")),
         "{:?}",
         seen.events
     );