Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions crates/tinyhivemind-driver/src/conduct/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:

Expand All @@ -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.
7 changes: 5 additions & 2 deletions crates/tinyhivemind-driver/src/conduct/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)),
}
Expand Down
72 changes: 60 additions & 12 deletions crates/tinyhivemind-driver/src/conduct/steps.rs
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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.
Expand All @@ -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 {
Expand All @@ -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 {
Expand All @@ -106,16 +136,22 @@ pub enum Event {
seat: String,
/// Who took it.
to: Vec<String>,
/// The broadcast row.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

priority critical critique confident

Update every event constructor for the new required fields

Adding at as a required field to event variants changes every Event::Broadcast, Unplaced, CompletedByBroadcast, Handoff, Refused, Discharged, and Concluded construction site. The unchanged conductor code still constructs these variants using their previous field sets, so the crate will fail to compile until each caller supplies the corresponding sequence (or the field is made optional/defaulted).

[RULE] build-break ·

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Every construction site is updated in this commit: the seven sites in conduct/wave.rs (Broadcast, Unplaced, CompletedByBroadcast, Handoff, both Refused arms and Discharged, plus Concluded), and the example's matches take ... The Rust job on this head compiles the crate and the example, so no site was missed.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Resolved — the reply explains why it is not a problem (advisory), as of 5d16143.

If this is wrong, reopen the conversation and say so; the finding will be re-raised on the next push if it still reproduces.

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 {
Expand All @@ -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 {
Expand All @@ -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),
}
70 changes: 67 additions & 3 deletions crates/tinyhivemind-driver/src/conduct/test/conversations.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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!(
Expand Down Expand Up @@ -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
);
Expand Down Expand Up @@ -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");

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

priority medium critique confident

Preserve the child until its conclusion commit succeeds

The failed conductor.committed(sequence) consumes the outstanding commit, and the conductor still removes the conversation before the driver's fold can succeed. As a result, this follow-up wave has no child left to retry or conclude, so the new regression test fails (or the conversation is permanently lost). Apply the conclusion first and remove the child only after the transition succeeds, preserving the outstanding conversation when the fold rejects the supplied sequence.

[RULE] state-preservation-on-error ·

assert!(
later
.events
.iter()
.any(|event| matches!(event, Event::Concluded { root: at, .. } if *at == root)),
"{:?}",
later.events
);
assert_eq!(conductor.conversations(), 1);
}
8 changes: 4 additions & 4 deletions crates/tinyhivemind-driver/src/conduct/test/desk.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
);
Expand Down
Loading
Loading