From 934d6aaa3015fa184d65c699cefb50f39ff1e4c1 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 22 Sep 2026 21:28:34 +0530 Subject: [PATCH 1/3] Give the conductor's steps wire forms, and link rows and events A host journals the conductor's commits and events and streams them to whatever draws the desk, so they now have pinned serde forms: internally tagged and snake_case, like Utterance and the driver's own payloads. Two links a desk needs were missing. A commit now names the conversation it belongs to -- a row said inside one, desk work lifted out of one, and the row that concludes it to its asker -- so an agent-to-agent exchange can be shown whole wherever its rows landed. And the events about a row now carry that row's sequence, with a handoff carrying the broadcast it came from, so a host attaches them to the row instead of inferring it from order. Co-Authored-By: Claude Opus 5 --- .../tinyhivemind-driver/src/conduct/README.md | 17 +- crates/tinyhivemind-driver/src/conduct/mod.rs | 7 +- .../tinyhivemind-driver/src/conduct/steps.rs | 72 +++++- .../src/conduct/test/conversations.rs | 6 +- .../src/conduct/test/desk.rs | 8 +- .../src/conduct/test/links.rs | 226 +++++++++++++++++ .../src/conduct/test/mod.rs | 2 + .../src/conduct/test/support.rs | 5 +- .../src/conduct/test/wire.rs | 237 ++++++++++++++++++ .../tinyhivemind-driver/src/conduct/wave.rs | 30 ++- .../tinyhivemind-driver/src/driver/brief.rs | 7 +- examples/openhuman/src/bin/conducted.rs | 15 +- 12 files changed, 595 insertions(+), 37 deletions(-) create mode 100644 crates/tinyhivemind-driver/src/conduct/test/links.rs create mode 100644 crates/tinyhivemind-driver/src/conduct/test/wire.rs diff --git a/crates/tinyhivemind-driver/src/conduct/README.md b/crates/tinyhivemind-driver/src/conduct/README.md index 2c82558c..a60a803d 100644 --- a/crates/tinyhivemind-driver/src/conduct/README.md +++ b/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. diff --git a/crates/tinyhivemind-driver/src/conduct/mod.rs b/crates/tinyhivemind-driver/src/conduct/mod.rs index deb865fd..39331a7a 100644 --- a/crates/tinyhivemind-driver/src/conduct/mod.rs +++ b/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)), } diff --git a/crates/tinyhivemind-driver/src/conduct/steps.rs b/crates/tinyhivemind-driver/src/conduct/steps.rs index 5878b628..96a3f0ff 100644 --- a/crates/tinyhivemind-driver/src/conduct/steps.rs +++ b/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, /// On the open desk, the one seat it reaches; `None` reaches every seat. pub only_for: Option, + /// 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, + /// 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, + /// 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,8 +168,11 @@ 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, @@ -141,11 +180,15 @@ pub enum Event { thread: Option, /// 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), } diff --git a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs index ca66792d..7b68990c 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs +++ b/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 ); diff --git a/crates/tinyhivemind-driver/src/conduct/test/desk.rs b/crates/tinyhivemind-driver/src/conduct/test/desk.rs index 7672876a..91a945f7 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/desk.rs +++ b/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 ); diff --git a/crates/tinyhivemind-driver/src/conduct/test/links.rs b/crates/tinyhivemind-driver/src/conduct/test/links.rs new file mode 100644 index 00000000..61ac8365 --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/links.rs @@ -0,0 +1,226 @@ +//! What ties a row to its conversation and an event to its row, as a host +//! reads them back to draw the desk. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use super::support::{ + ClarifyRouter, Journal, ask, broadcast, complete, door, hive, policy, post, two_seat, wave, +}; +use crate::CompletionDriver; +use crate::conduct::{ConductPolicy, Conductor, Event, Refusal}; +use crate::driver::BroadcastRouting; +use tinyhivemind::Sequence; +use tinyhivemind::speech::Utterance; + +#[test] +fn every_row_of_a_conversation_carries_its_root_wherever_it_lands() { + let hive = hive(&["one", "two", "three"]); + let driver = CompletionDriver::new(&hive, 4).expect("driver"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two", "three"], &["one"], &journal), + ) + .expect("opens"); + + let opened = wave( + &mut conductor, + &journal, + &[( + "one", + vec![post("looking into it"), ask("two", "which port?")], + )], + ) + .expect("wave"); + let (posted_at, posted) = &opened.commits[0]; + assert_eq!( + posted.conversation, None, + "a desk row belongs to no conversation" + ); + assert_eq!(*posted_at, Sequence(2)); + let (root, asking) = &opened.commits[1]; + assert!(matches!(asking.utterance, Utterance::Ask { .. })); + assert_eq!( + asking.conversation, None, + "the ask is a desk row; its own sequence is the conversation" + ); + assert!(opened.events.iter().any(|event| matches!( + event, + Event::Asked { root: at, .. } if at == root + ))); + + let talked = wave( + &mut conductor, + &journal, + &[( + "two", + vec![ + post("checking the config"), + broadcast("three should check the logs"), + complete("port 8080"), + ], + )], + ) + .expect("wave"); + for (_, commit) in &talked.commits { + assert_eq!( + commit.conversation, + Some(*root), + "{:?} belongs to the conversation", + commit.utterance + ); + } + let lifted = talked + .commits + .iter() + .find(|(_, commit)| commit.utterance.broadcasting()) + .map(|(_, commit)| commit) + .expect("the broadcast was committed"); + assert_eq!(lifted.thread, None, "desk work lands on the desk"); + let in_thread: Vec<_> = talked + .commits + .iter() + .filter(|(_, commit)| commit.thread == Some(*root)) + .collect(); + assert_eq!( + in_thread.len(), + 2, + "the post and the answer are in the thread" + ); + let (concluded_at, conclusion) = talked + .commits + .iter() + .find(|(_, commit)| matches!(commit.utterance, Utterance::Dm { .. })) + .expect("the conclusion reached the asker"); + assert_eq!(conclusion.thread, None); + assert_eq!(conclusion.only_for.as_deref(), Some("one")); + assert!(talked.events.iter().any(|event| matches!( + event, + Event::Concluded { root: r, at, forced: false, .. } if r == root && at == concluded_at + ))); +} + +#[test] +fn a_broadcast_event_names_the_row_it_placed_and_the_handoff_its_origin() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4) + .expect("driver") + .with_queue_depth(2) + .expect("depth"); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: None, + reasoning: None, + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = Conductor::open( + &driver, + routing, + ConductPolicy::default(), + door(&["one", "two"], &["one", "two"], &journal), + ) + .expect("opens"); + let placed = wave( + &mut conductor, + &journal, + &[("one", vec![broadcast("two: also check the cache")])], + ) + .expect("wave"); + let (broadcast_at, _) = placed.commits[0]; + for event in &placed.events { + match event { + Event::Broadcast { at, .. } | Event::CompletedByBroadcast { at, .. } => { + assert_eq!(*at, broadcast_at); + } + _ => {} + } + } + assert!( + placed + .events + .iter() + .any(|event| matches!(event, Event::CompletedByBroadcast { .. })) + ); + let handed = wave(&mut conductor, &journal, &[("two", vec![complete("mine")])]).expect("wave"); + assert!(handed.events.iter().any(|event| matches!( + event, + Event::Handoff { origin, .. } if *origin == broadcast_at + ))); +} + +#[test] +fn a_refused_unplaced_or_discharged_row_is_named_by_its_sequence() { + let hive = hive(&["one", "two"]); + let driver = CompletionDriver::new(&hive, 4) + .expect("driver") + .with_broadcast_budget(Some(1)); + let route_policy = policy(1); + let routing = BroadcastRouting { + primary: Some(&ClarifyRouter), + reasoning: Some(&ClarifyRouter), + policy: &route_policy, + roster_version: 1, + thread_context: &[], + }; + let journal = Journal::default(); + let mut conductor = two_seat(&driver, routing, ConductPolicy::default(), &journal); + let seen = wave( + &mut conductor, + &journal, + &[("one", vec![broadcast("first"), broadcast("second")])], + ) + .expect("wave"); + let first = seen.commits[0].0; + let second = seen.commits[1].0; + assert!(seen.events.iter().any(|event| matches!( + event, + Event::Unplaced { at, .. } if *at == first + ))); + assert!(seen.events.iter().any(|event| matches!( + event, + Event::Discharged { at, .. } if *at == second + ))); + + // A completion refused while a conversation is open names its own row. + let hive = super::support::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 seen = wave( + &mut conductor, + &journal, + &[("one", vec![ask("two", "?"), complete("too soon")])], + ) + .expect("wave"); + let completion_at = seen + .commits + .iter() + .find(|(_, commit)| matches!(commit.utterance, Utterance::CompleteEpisode { .. })) + .map(|(at, _)| *at) + .expect("the completion was committed"); + assert!(seen.events.iter().any(|event| matches!( + event, + Event::Refused { why: Refusal::AwaitingReply { .. }, at, .. } if *at == completion_at + ))); +} diff --git a/crates/tinyhivemind-driver/src/conduct/test/mod.rs b/crates/tinyhivemind-driver/src/conduct/test/mod.rs index 3ce1eba6..93041eec 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/mod.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/mod.rs @@ -4,4 +4,6 @@ mod conversations; mod desk; mod door; +mod links; mod support; +mod wire; diff --git a/crates/tinyhivemind-driver/src/conduct/test/support.rs b/crates/tinyhivemind-driver/src/conduct/test/support.rs index c8aa92ce..46a56153 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/support.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/support.rs @@ -12,7 +12,7 @@ use tinyhivemind_embed::{ RouterFuture, RoutingEvaluation, RoutingPolicy, RoutingRequest, }; -use crate::conduct::{ConductPolicy, Conductor, Door, Event, Step, Turn}; +use crate::conduct::{Commit, ConductPolicy, Conductor, Door, Event, Step, Turn}; use crate::driver::BroadcastRouting; use crate::test_support::Seat; use crate::{AgentBinding, BoundHive, CompletionDriver, Error, HiveGraph}; @@ -188,6 +188,8 @@ impl Journal { pub(super) struct Wave { pub(super) turns: Vec, pub(super) events: Vec, + /// Every commit the wave handed out, with the sequence it was given. + pub(super) commits: Vec<(Sequence, Commit)>, } /// One wave: nudges, turns, the scripted calls each seat makes, and every @@ -224,6 +226,7 @@ pub(super) fn wave( commit.only_for.clone(), ); run(conductor.committed(sequence))?; + seen.commits.push((sequence, commit.clone())); continue; } take(step, journal, &mut seen); diff --git a/crates/tinyhivemind-driver/src/conduct/test/wire.rs b/crates/tinyhivemind-driver/src/conduct/test/wire.rs new file mode 100644 index 00000000..c40601ea --- /dev/null +++ b/crates/tinyhivemind-driver/src/conduct/test/wire.rs @@ -0,0 +1,237 @@ +//! The wire forms: what a host journals and streams, pinned exactly. + +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use serde_json::{Value, json}; +use tinyhivemind::Sequence; +use tinyhivemind::speech::Utterance; + +use crate::conduct::steps::Kind; +use crate::conduct::{Commit, Event, Note, Refusal, Step, Turn}; +use crate::driver::Channel; + +/// Every field in `required` must be present: a payload missing one fails to +/// decode rather than decoding to something the conductor never said. +fn rejects_missing(value: &Value, required: &[&str]) { + for field in required { + let mut payload = value.as_object().expect("an object").clone(); + payload.remove(*field); + assert!( + serde_json::from_value::(payload.into()).is_err(), + "missing {field} must be rejected" + ); + } +} + +fn round_trips(value: &T) +where + T: serde::Serialize + serde::de::DeserializeOwned + PartialEq + std::fmt::Debug, +{ + let wire = serde_json::to_value(value).expect("serializes"); + let back: T = serde_json::from_value(wire).expect("deserializes"); + assert_eq!(&back, value); +} + +#[test] +fn a_turn_names_its_seat_channel_and_watermark() { + let thread = Turn { + seat: "two".into(), + channel: Channel::Thread { + root: Sequence(4), + other: "one".into(), + opened_it: false, + }, + since: Sequence(4), + }; + let wire = serde_json::to_value(&thread).expect("serializes"); + assert_eq!( + wire, + json!({ + "seat": "two", + "channel": {"kind": "thread", "root": 4, "other": "one", "opened_it": false}, + "since": 4 + }) + ); + rejects_missing::(&wire, &["seat", "channel", "since"]); + round_trips(&thread); + + let desk = Turn { + seat: "one".into(), + channel: Channel::Desk, + since: Sequence(0), + }; + assert_eq!( + serde_json::to_value(&desk).expect("serializes")["channel"], + json!({"kind": "desk"}) + ); + round_trips(&desk); +} + +#[test] +fn a_note_is_tagged_as_a_step_with_its_fields_beside_the_tag() { + let note = Step::Note(Note { + body: "you hold open work".into(), + thread: None, + only_for: Some("one".into()), + }); + assert_eq!( + serde_json::to_value(¬e).expect("serializes"), + json!({ + "step": "note", + "body": "you hold open work", + "thread": null, + "only_for": "one" + }) + ); + round_trips(¬e); +} + +#[test] +fn a_commit_carries_its_conversation_and_an_opaque_purpose() { + let lifted = Commit { + author: "two".into(), + utterance: Utterance::Broadcast { + message: "three should check the logs".into(), + }, + thread: None, + only_for: None, + conversation: Some(Sequence(4)), + kind: Kind::Desk, + }; + let wire = serde_json::to_value(Step::Commit(lifted.clone())).expect("serializes"); + assert_eq!( + wire, + json!({ + "step": "commit", + "author": "two", + "utterance": {"kind": "broadcast", "message": "three should check the logs"}, + "thread": null, + "only_for": null, + "conversation": 4, + "purpose": {"kind": "desk"} + }) + ); + let bare = serde_json::to_value(&lifted).expect("serializes"); + rejects_missing::(&bare, &["author", "utterance", "purpose"]); + round_trips(&Step::Commit(lifted)); + + for kind in [ + Kind::Thread { root: Sequence(4) }, + Kind::Desk, + Kind::Conclusion { + root: Sequence(4), + forced: true, + }, + Kind::Discharge, + ] { + round_trips(&Commit { + author: "one".into(), + utterance: Utterance::CompleteEpisode { + message: "done".into(), + }, + thread: Some(Sequence(4)), + only_for: None, + conversation: Some(Sequence(4)), + kind, + }); + } +} + +#[test] +fn an_event_is_tagged_by_kind_inside_its_step() { + let asked = Step::Event(Event::Asked { + seat: "one".into(), + askee: "two".into(), + root: Sequence(4), + }); + assert_eq!( + serde_json::to_value(&asked).expect("serializes"), + json!({"step": "event", "kind": "asked", "seat": "one", "askee": "two", "root": 4}) + ); + let refused = Event::Refused { + seat: "one".into(), + thread: None, + why: Refusal::AwaitingReply { + waiting_on: vec!["two".into()], + }, + at: Sequence(5), + }; + let wire = serde_json::to_value(&refused).expect("serializes"); + assert_eq!( + wire, + json!({ + "kind": "refused", + "seat": "one", + "thread": null, + "why": {"kind": "awaiting_reply", "waiting_on": ["two"]}, + "at": 5 + }) + ); + rejects_missing::(&wire, &["kind", "seat", "why", "at"]); +} + +#[test] +fn every_event_and_refusal_survives_the_wire() { + let events = [ + Event::Nudged { + seat: "one".into(), + thread: Some(Sequence(4)), + }, + Event::Broadcast { + seat: "one".into(), + to: vec!["two".into()], + at: Sequence(3), + }, + Event::Unplaced { + seat: "one".into(), + at: Sequence(3), + }, + Event::CompletedByBroadcast { + seat: "one".into(), + at: Sequence(3), + }, + Event::Asked { + seat: "one".into(), + askee: "two".into(), + root: Sequence(4), + }, + Event::Handoff { + to: "two".into(), + from: "one".into(), + origin: Sequence(3), + }, + Event::Refused { + seat: "two".into(), + thread: Some(Sequence(4)), + why: Refusal::NotYetShown, + at: Sequence(6), + }, + Event::Refused { + seat: "two".into(), + thread: None, + why: Refusal::Undelivered { + assigned_at: Sequence(5), + }, + at: Sequence(6), + }, + Event::Discharged { + seat: "one".into(), + at: Sequence(7), + }, + Event::Concluded { + root: Sequence(4), + asker: "one".into(), + askee: "two".into(), + forced: false, + at: Sequence(8), + }, + ]; + for event in events { + round_trips(&Step::Event(event.clone())); + round_trips(&event); + } + assert_eq!( + serde_json::to_value(Refusal::NotYetShown).expect("serializes"), + json!({"kind": "not_yet_shown"}) + ); +} diff --git a/crates/tinyhivemind-driver/src/conduct/wave.rs b/crates/tinyhivemind-driver/src/conduct/wave.rs index 79ec69f2..9d00710f 100644 --- a/crates/tinyhivemind-driver/src/conduct/wave.rs +++ b/crates/tinyhivemind-driver/src/conduct/wave.rs @@ -47,8 +47,9 @@ pub(super) struct Wave { force_conclusions: bool, /// What thread turns said: `(root, seat, utterance)`. pub(super) thread: Vec<(Sequence, String, Utterance)>, - /// What desk turns said, and what thread turns said to the desk. - pub(super) desk: Vec<(String, Utterance)>, + /// What desk turns said, and what thread turns said to the desk: + /// `(seat, utterance, the conversation it was lifted out of)`. + pub(super) desk: Vec<(String, Utterance, Option)>, /// Steps ready for the host, notes and events. steps: VecDeque, /// Commits waiting for the host, in order. @@ -112,13 +113,14 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { self.wave.phase = Phase::Desk; } Phase::Desk => { - for (seat, utterance) in std::mem::take(&mut self.wave.desk) { + for (seat, utterance, conversation) in std::mem::take(&mut self.wave.desk) { let only_for = utterance.asks().map(str::to_owned); self.wave.commits.push_back(Commit { author: seat, utterance, thread: None, only_for, + conversation, kind: Kind::Desk, }); } @@ -156,7 +158,8 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { utterance, thread: Some(root), only_for: None, - kind: Kind::Thread(root), + conversation: Some(root), + kind: Kind::Thread { root }, }); } @@ -215,6 +218,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { }, thread: None, only_for: Some(child.asker.clone()), + conversation: Some(root), kind: Kind::Conclusion { root, forced }, }); } @@ -238,7 +242,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { utterance: commit.utterance.clone(), }; match commit.kind { - Kind::Thread(root) => self.commit_thread(root, committed).await, + Kind::Thread { root } => self.commit_thread(root, committed).await, Kind::Desk => self.commit_desk(committed).await, Kind::Conclusion { root, forced } => { self.commit_conclusion(root, forced, committed).await @@ -259,6 +263,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { return Ok(()); }; let seat = committed.author_id.clone(); + let at = committed.sequence; let said = committed.utterance.message().to_owned(); match self .driver @@ -276,6 +281,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { seat, thread: Some(root), why: Refusal::NotYetShown, + at, }), Err(error) => return Err(error), } @@ -323,6 +329,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { seat, thread: None, why: Refusal::AwaitingReply { waiting_on }, + at: sequence, }); } Err(Error::UndeliveredAssignment { assigned_at, .. }) => { @@ -339,11 +346,15 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { seat, thread: None, why: Refusal::Undelivered { assigned_at }, + at: sequence, }); } Err(Error::BudgetSpent { .. }) => { self.discharged += 1; - self.wave.event(Event::Discharged { seat: seat.clone() }); + self.wave.event(Event::Discharged { + seat: seat.clone(), + at: sequence, + }); // Before whatever else the wave said: the seat keeps the work // now, so a later row from it applies to that. self.wave.commits.push_front(Commit { @@ -353,6 +364,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { }, thread: None, only_for: None, + conversation: None, kind: Kind::Discharge, }); } @@ -372,6 +384,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { self.wave.event(Event::Broadcast { seat: said.seat.clone(), to: agent_ids, + at: said.sequence, }); } // For an ask, this is the signal to open the conversation: a @@ -386,6 +399,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { self.wave.event(Event::Handoff { to: agent_id.clone(), from: handoff.from.clone(), + origin: handoff.origin, }); self.wave.note( format!("handoff from @{}: {}", handoff.from, handoff.body), @@ -398,11 +412,13 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { if said.is_broadcast && said.held && !holds(&transition.state, &said.seat) { self.wave.event(Event::CompletedByBroadcast { seat: said.seat.clone(), + at: said.sequence, }); } if said.is_broadcast && !routed { self.wave.event(Event::Unplaced { seat: said.seat.clone(), + at: said.sequence, }); self.wave.note( "nobody on this desk can take that; the work stays with you. Do what you can \ @@ -443,6 +459,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { let Some(child) = self.children.remove(&root) else { return Ok(()); }; + let at = committed.sequence; let transition = self .driver .apply_committed(&self.state, committed, None) @@ -453,6 +470,7 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { asker: child.asker.clone(), askee: child.askee.clone(), forced, + at, }); self.concluded.push(Concluded { root, diff --git a/crates/tinyhivemind-driver/src/driver/brief.rs b/crates/tinyhivemind-driver/src/driver/brief.rs index d4243d06..64f96c70 100644 --- a/crates/tinyhivemind-driver/src/driver/brief.rs +++ b/crates/tinyhivemind-driver/src/driver/brief.rs @@ -11,6 +11,7 @@ //! Nothing here reads storage. The host passes the rows a seat may see and //! the conversations it was part of; the brief adds only what the state holds. +use serde::{Deserialize, Serialize}; use tinyhivemind::Sequence; use tinyhivemind::speech::ToolSpec; @@ -18,7 +19,11 @@ use super::DriverState; use super::ledger::open_assignment; /// Where a turn runs. -#[derive(Clone, Debug, Eq, PartialEq)] +/// +/// On the wire, tagged by `kind`: `{"kind":"desk"}` or +/// `{"kind":"thread","root":7,"other":"two","opened_it":false}`. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] pub enum Channel { /// The open desk. Desk, diff --git a/examples/openhuman/src/bin/conducted.rs b/examples/openhuman/src/bin/conducted.rs index b3482f0d..d3af0dc9 100644 --- a/examples/openhuman/src/bin/conducted.rs +++ b/examples/openhuman/src/bin/conducted.rs @@ -941,17 +941,19 @@ fn log(event: &Event) { seat, thread: Some(root), } => eprintln!("[nudged] @{seat} in thread {}", root.0), - Event::Broadcast { seat, to } => println!("[broadcast] @{seat} -> {}", to.join(", ")), - Event::Unplaced { seat } => { + Event::Broadcast { seat, to, .. } => println!("[broadcast] @{seat} -> {}", to.join(", ")), + Event::Unplaced { seat, .. } => { println!("[unplaced] @{seat}'s broadcast fits no seat; it keeps the work"); } - Event::CompletedByBroadcast { seat } => eprintln!("[completed] @{seat} by its broadcast"), + Event::CompletedByBroadcast { seat, .. } => eprintln!("[completed] @{seat} by its broadcast"), Event::Asked { seat, askee, root } => println!( "[ask] @{seat} opened a conversation with @{askee} (thread {})", root.0 ), - Event::Handoff { to, from } => println!("[handoff] -> @{to} (queued from @{from})"), - Event::Refused { seat, thread, why } => { + Event::Handoff { to, from, .. } => println!("[handoff] -> @{to} (queued from @{from})"), + Event::Refused { + seat, thread, why, .. + } => { let where_ = thread.map_or(String::new(), |root| format!(" in thread {}", root.0)); let reason = match why { Refusal::AwaitingReply { waiting_on } => { @@ -965,7 +967,7 @@ fn log(event: &Event) { }; eprintln!("[refused] @{seat}{where_}: {reason}"); } - Event::Discharged { seat } => { + Event::Discharged { seat, .. } => { eprintln!("[refused] @{seat} has spent its broadcast budget; it keeps the work"); } Event::Concluded { @@ -973,6 +975,7 @@ fn log(event: &Event) { asker, askee, forced, + .. } => println!( "[concluded] thread {} between @{asker} and @{askee}{}", root.0, From ff87c9acf3ad8cf4888f5dde0e9721debfd617ad Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 22 Sep 2026 22:01:54 +0530 Subject: [PATCH 2/3] Keep a conversation open when the fold refuses its conclusion The child was removed before its conclusion was applied, so a conclusion the fold refused left the desk unchanged and the conversation gone, with no way to conclude it later. The fold goes first now; a refused conclusion leaves the conversation open, and the next wave concludes it at a row the host gives properly. Co-Authored-By: Claude Fable 5.1 --- .../src/conduct/test/conversations.rs | 69 +++++++++++++++++++ .../tinyhivemind-driver/src/conduct/wave.rs | 9 ++- 2 files changed, 76 insertions(+), 2 deletions(-) diff --git a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs index 7b68990c..64a086e6 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs @@ -423,3 +423,72 @@ 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; + loop { + match conductor.step() { + Ok(None) => break, + Ok(Some(Step::Commit(commit))) => { + 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; + } + } + Ok(Some(_)) => {} + Err(error) => panic!("{error}"), + } + } + 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); +} diff --git a/crates/tinyhivemind-driver/src/conduct/wave.rs b/crates/tinyhivemind-driver/src/conduct/wave.rs index 9d00710f..6293e0cf 100644 --- a/crates/tinyhivemind-driver/src/conduct/wave.rs +++ b/crates/tinyhivemind-driver/src/conduct/wave.rs @@ -456,15 +456,20 @@ impl<'a, A: BoundAgent> Conductor<'a, A> { forced: bool, committed: CommittedUtterance, ) -> Result<()> { - let Some(child) = self.children.remove(&root) else { + if !self.children.contains_key(&root) { return Ok(()); - }; + } let at = committed.sequence; + // The fold first: a conclusion it refuses leaves the conversation + // open, to be concluded again on a later wave, rather than gone. let transition = self .driver .apply_committed(&self.state, committed, None) .await?; self.state = transition.state; + let Some(child) = self.children.remove(&root) else { + return Ok(()); + }; self.wave.event(Event::Concluded { root, asker: child.asker.clone(), From 5d16143dae3670805a40b804c8c0bf691c95338c Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 22 Sep 2026 22:02:58 +0530 Subject: [PATCH 3/3] Walk the refused-conclusion test's steps without a panic arm Co-Authored-By: Claude Fable 5.1 --- .../src/conduct/test/conversations.rs | 33 ++++++++----------- 1 file changed, 14 insertions(+), 19 deletions(-) diff --git a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs index 64a086e6..593d5d3a 100644 --- a/crates/tinyhivemind-driver/src/conduct/test/conversations.rs +++ b/crates/tinyhivemind-driver/src/conduct/test/conversations.rs @@ -454,26 +454,21 @@ fn a_conclusion_the_fold_refuses_leaves_the_conversation_to_conclude_later() { } } let mut refused = false; - loop { - match conductor.step() { - Ok(None) => break, - Ok(Some(Step::Commit(commit))) => { - 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; - } + 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; } - Ok(Some(_)) => {} - Err(error) => panic!("{error}"), } } assert!(refused, "a reused sequence is refused by the fold");