diff --git a/src/bus.rs b/src/bus.rs new file mode 100644 index 0000000..8f9e585 --- /dev/null +++ b/src/bus.rs @@ -0,0 +1,196 @@ +//! The turn bus: fans turn events out to every registered listener. + +use std::ops::ControlFlow; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::Sender; + +use ratatui::text::Text; + +use crate::campaign::Campaign; +use crate::play::worker::TurnEvent; + +/// What one listener saw and wants to happen next. +pub enum Flow { + Continue, + Break, +} + +/// A thing that reacts to turn events as they happen. +/// Every method has a default no-op implementation. +pub trait Listener { + fn delta(&mut self, _text: &str) -> Flow { + Flow::Continue + } + fn tool(&mut self, _display: Option<&Text<'static>>) -> Flow { + Flow::Continue + } + fn done(&mut self, _reply: &str) {} + fn cancelled(&mut self, _partial: &str) {} + fn failed(&mut self, _error: &str) {} +} + +/// A bus that fans turn events out to every listener. +#[derive(Default)] +pub struct TurnBus { + listeners: Vec>, +} + +impl TurnBus { + pub fn new() -> Self { + Self::default() + } + + pub fn add(&mut self, listener: Box) { + self.listeners.push(listener); + } + + pub fn delta(&mut self, text: &str) -> ControlFlow<()> { + for listener in &mut self.listeners { + if matches!(listener.delta(text), Flow::Break) { + return ControlFlow::Break(()); + } + } + ControlFlow::Continue(()) + } + + pub fn tool(&mut self, display: Option<&Text<'static>>) -> ControlFlow<()> { + for listener in &mut self.listeners { + if matches!(listener.tool(display), Flow::Break) { + return ControlFlow::Break(()); + } + } + ControlFlow::Continue(()) + } + + pub fn done(&mut self, reply: &str) { + for listener in &mut self.listeners { + listener.done(reply); + } + } + + pub fn cancelled(&mut self, partial: &str) { + for listener in &mut self.listeners { + listener.cancelled(partial); + } + } + + pub fn failed(&mut self, error: &str) { + for listener in &mut self.listeners { + listener.failed(error); + } + } +} + +/// A listener that sends turn events to the screen over a channel. +pub struct ScreenListener { + sender: Sender, + cancel: Arc, +} + +impl ScreenListener { + pub fn new(sender: Sender, cancel: Arc) -> Self { + Self { sender, cancel } + } +} + +impl Listener for ScreenListener { + fn delta(&mut self, text: &str) -> Flow { + // Send the event first (ignoring send failures), then check + // the cancel flag, matching the old callback behavior where + // the delta was sent before checking cancel. + let _ = self.sender.send(TurnEvent::Delta(text.to_string())); + if self.cancel.load(Ordering::Relaxed) { + return Flow::Break; + } + Flow::Continue + } + + fn tool(&mut self, display: Option<&Text<'static>>) -> Flow { + if let Some(text) = display { + // Send the event first, then check cancel. + let _ = self.sender.send(TurnEvent::Tool(text.clone())); + } + if self.cancel.load(Ordering::Relaxed) { + return Flow::Break; + } + Flow::Continue + } + + fn done(&mut self, reply: &str) { + let _ = self.sender.send(TurnEvent::Done(reply.to_string())); + } + + fn cancelled(&mut self, partial: &str) { + let _ = self.sender.send(TurnEvent::Cancelled(partial.to_string())); + } + + fn failed(&mut self, error: &str) { + let _ = self.sender.send(TurnEvent::Failed(error.to_string())); + } +} + +/// A listener that records the turn to the campaign transcript. +pub struct CampaignListener { + campaign: Option, + input: String, + narration: String, +} + +impl CampaignListener { + pub fn new(campaign: Option, input: String) -> Self { + Self { + campaign, + input, + narration: String::new(), + } + } +} + +impl Listener for CampaignListener { + fn delta(&mut self, text: &str) -> Flow { + self.narration.push_str(text); + Flow::Continue + } + + fn done(&mut self, reply: &str) { + // Ensure any delta text that arrived is included (and also append + // the full reply so nothing is lost even if done() is the only + // way we see the final text). + let text = if self.narration.is_empty() { + reply.to_string() + } else { + self.narration.clone() + }; + + let Some(campaign) = &self.campaign else { + return; + }; + + // Record the player's line. The clock at turn-start is captured + // in new(), before any `mark` mid-turn advances it. + let start_time = campaign.current_time().ok(); + if let Some(time) = start_time { + let _ = campaign.append_player(time, &self.input); + } + + // Record narration. The clock may have advanced via mark() + // mid-turn, so we read it again at done() time. + let now = campaign.current_time().ok().or(start_time); + if let (Some(time), true) = (now, !text.trim().is_empty()) { + let _ = campaign.append_narration(time, &text); + } + } + + fn cancelled(&mut self, _partial: &str) { + // Don't record cancelled turns + } + + fn failed(&mut self, _error: &str) { + // Don't record failed turns + } +} + +#[cfg(test)] +#[path = "bus_tests.rs"] +mod tests; diff --git a/src/bus_tests.rs b/src/bus_tests.rs new file mode 100644 index 0000000..c703a63 --- /dev/null +++ b/src/bus_tests.rs @@ -0,0 +1,365 @@ +//! Tests for the turn bus and its listeners. + +use super::*; +use crate::play::worker::TurnEvent; +use std::sync::atomic::AtomicBool; +use std::sync::mpsc; + +/// A listener that collects events for assertions. +#[derive(Clone)] +struct CollectingListener { + inner: std::sync::Arc>, +} + +#[derive(Default)] +struct Collected { + deltas: Vec, + tools: Vec>, + done: Option, + cancelled: Option, + failed: Option, + break_on_delta: bool, + break_on_tool: bool, +} + +impl CollectingListener { + fn new() -> Self { + Self { + inner: std::sync::Arc::new(std::sync::Mutex::new(Collected::default())), + } + } + + fn break_on_delta(self) -> Self { + self.inner.lock().unwrap().break_on_delta = true; + self + } + + fn break_on_tool(self) -> Self { + self.inner.lock().unwrap().break_on_tool = true; + self + } + + fn deltas(&self) -> Vec { + self.inner.lock().unwrap().deltas.clone() + } + + fn done(&self) -> Option { + self.inner.lock().unwrap().done.clone() + } + + fn cancelled(&self) -> Option { + self.inner.lock().unwrap().cancelled.clone() + } + + fn failed(&self) -> Option { + self.inner.lock().unwrap().failed.clone() + } +} + +impl Listener for CollectingListener { + fn delta(&mut self, text: &str) -> Flow { + let mut state = self.inner.lock().unwrap(); + state.deltas.push(text.to_string()); + if state.break_on_delta { + Flow::Break + } else { + Flow::Continue + } + } + + fn tool(&mut self, display: Option<&Text<'static>>) -> Flow { + if let Some(text) = display { + let mut state = self.inner.lock().unwrap(); + state.tools.push(text.clone()); + if state.break_on_tool { + return Flow::Break; + } + } + Flow::Continue + } + + fn done(&mut self, reply: &str) { + self.inner.lock().unwrap().done = Some(reply.to_string()); + } + + fn cancelled(&mut self, partial: &str) { + self.inner.lock().unwrap().cancelled = Some(partial.to_string()); + } + + fn failed(&mut self, error: &str) { + self.inner.lock().unwrap().failed = Some(error.to_string()); + } +} + +#[test] +fn empty_bus_delta_returns_continue() { + let mut bus = TurnBus::new(); + assert_eq!(bus.delta("hello"), ControlFlow::Continue(())); +} + +#[test] +fn empty_bus_tool_returns_continue() { + let mut bus = TurnBus::new(); + assert_eq!(bus.tool(None), ControlFlow::Continue(())); +} + +#[test] +fn empty_bus_done_does_not_panic() { + let mut bus = TurnBus::new(); + bus.done("hello"); +} + +#[test] +fn empty_bus_cancelled_does_not_panic() { + let mut bus = TurnBus::new(); + bus.cancelled("partial"); +} + +#[test] +fn empty_bus_failed_does_not_panic() { + let mut bus = TurnBus::new(); + bus.failed("error"); +} + +#[test] +fn single_listener_receives_delta() { + let mut bus = TurnBus::new(); + let collector = CollectingListener::new(); + bus.add(Box::new(collector.clone())); + + let _ = bus.delta("You wake."); + + assert_eq!(collector.deltas(), vec!["You wake."]); +} + +#[test] +fn multiple_listeners_all_receive_delta() { + let mut bus = TurnBus::new(); + let c1 = CollectingListener::new(); + let c2 = CollectingListener::new(); + bus.add(Box::new(c1.clone())); + bus.add(Box::new(c2.clone())); + + let _ = bus.delta("Hello"); + + assert_eq!(c1.deltas(), vec!["Hello"]); + assert_eq!(c2.deltas(), vec!["Hello"]); +} + +#[test] +fn delta_break_stops_iteration() { + let mut bus = TurnBus::new(); + let c1 = CollectingListener::new().break_on_delta(); + let c2 = CollectingListener::new(); + bus.add(Box::new(c1.clone())); + bus.add(Box::new(c2.clone())); + + assert_eq!(bus.delta("stop"), ControlFlow::Break(())); + // c2 should not have received the event + assert!(c2.deltas().is_empty()); +} + +#[test] +fn tool_break_stops_iteration() { + let mut bus = TurnBus::new(); + let c1 = CollectingListener::new().break_on_tool(); + let c2 = CollectingListener::new(); + bus.add(Box::new(c1.clone())); + bus.add(Box::new(c2.clone())); + + let display: Text<'static> = Text::raw("rolled 20"); + assert_eq!(bus.tool(Some(&display)), ControlFlow::Break(())); + // c2 should not have received the event + assert!(c2.deltas().is_empty()); +} + +#[test] +fn done_fans_to_all_listeners() { + let mut bus = TurnBus::new(); + let c1 = CollectingListener::new(); + let c2 = CollectingListener::new(); + bus.add(Box::new(c1.clone())); + bus.add(Box::new(c2.clone())); + + bus.done("finished"); + + assert_eq!(c1.done(), Some("finished".to_string())); + assert_eq!(c2.done(), Some("finished".to_string())); +} + +#[test] +fn cancelled_fans_to_all_listeners() { + let mut bus = TurnBus::new(); + let c1 = CollectingListener::new(); + let c2 = CollectingListener::new(); + bus.add(Box::new(c1.clone())); + bus.add(Box::new(c2.clone())); + + bus.cancelled("partial"); + + assert_eq!(c1.cancelled(), Some("partial".to_string())); + assert_eq!(c2.cancelled(), Some("partial".to_string())); +} + +#[test] +fn failed_fans_to_all_listeners() { + let mut bus = TurnBus::new(); + let c1 = CollectingListener::new(); + let c2 = CollectingListener::new(); + bus.add(Box::new(c1.clone())); + bus.add(Box::new(c2.clone())); + + bus.failed("something broke"); + + assert_eq!(c1.failed(), Some("something broke".to_string())); + assert_eq!(c2.failed(), Some("something broke".to_string())); +} + +#[test] +fn screen_listener_sends_delta_events() { + let (sender, receiver) = mpsc::channel(); + let cancel = Arc::new(AtomicBool::new(false)); + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new(sender, cancel))); + + let _ = bus.delta("Hello"); + + assert_eq!( + receiver.recv().unwrap(), + TurnEvent::Delta("Hello".to_string()) + ); +} + +#[test] +fn screen_listener_cancel_flag_breaks_on_delta() { + let (sender, receiver) = mpsc::channel(); + let cancel = Arc::new(AtomicBool::new(true)); + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new(sender, cancel))); + + // The delta is still sent even when cancel is set (matching old + // callback behavior: send first, then check cancel). + assert_eq!(bus.delta("Hello"), ControlFlow::Break(())); + assert_eq!( + receiver.recv().unwrap(), + TurnEvent::Delta("Hello".to_string()) + ); +} + +#[test] +fn screen_listener_closed_channel_does_not_break() { + let (sender, receiver) = mpsc::channel::(); + drop(receiver); // close the receiver side + let cancel = Arc::new(AtomicBool::new(false)); + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new(sender, cancel))); + + // A closed channel does not cause a break (matching old callback + // behavior where channel send failures were silently ignored). + assert_eq!(bus.delta("Hello"), ControlFlow::Continue(())); +} + +#[test] +fn tool_without_display_does_not_send() { + let (sender, receiver) = mpsc::channel(); + let cancel = Arc::new(AtomicBool::new(false)); + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new(sender, cancel))); + + let _ = bus.tool(None); + + // Nothing should have been sent; the channel should be empty + assert!(receiver.try_recv().is_err()); +} + +#[test] +fn tool_with_display_sends_tool_event() { + let (sender, receiver) = mpsc::channel(); + let cancel = Arc::new(AtomicBool::new(false)); + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new(sender, cancel))); + + let display: Text<'static> = Text::raw("rolled 20"); + let _ = bus.tool(Some(&display)); + + assert!(matches!(receiver.recv().unwrap(), TurnEvent::Tool(_))); +} + +#[test] +fn campaign_listener_accumulates_narration_text() { + let mut bus = TurnBus::new(); + let listener = CampaignListener::new(None, "I sleep.".to_string()); + bus.add(Box::new(listener)); + + let _ = bus.delta("You "); + let _ = bus.delta("wake."); + + // No campaign to write to, but we verify it doesn't panic and the + // narration was accumulated (internal state verified via done). + bus.done("You wake."); +} + +#[test] +fn campaign_listener_done_falls_back_to_reply_when_narration_empty() { + // When no deltas were received, done() uses the reply as the text to + // record. With no campaign attached, the method just returns + // without panicking. + let mut bus = TurnBus::new(); + let listener = CampaignListener::new(None, "I sleep.".to_string()); + bus.add(Box::new(listener)); + + bus.done("You wake."); +} + +#[test] +fn campaign_listener_cancelled_is_a_noop() { + let mut bus = TurnBus::new(); + let listener = CampaignListener::new(None, "I sleep.".to_string()); + bus.add(Box::new(listener)); + + bus.cancelled("partial reply"); +} + +#[test] +fn campaign_listener_failed_is_a_noop() { + let mut bus = TurnBus::new(); + let listener = CampaignListener::new(None, "I sleep.".to_string()); + bus.add(Box::new(listener)); + + bus.failed("something broke"); +} + +#[test] +fn screen_listener_failed_sends_failed_event() { + let (sender, receiver) = mpsc::channel(); + let cancel = Arc::new(AtomicBool::new(false)); + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new(sender, cancel))); + + bus.failed("something broke"); + + assert_eq!( + receiver.recv().unwrap(), + TurnEvent::Failed("something broke".to_string()) + ); +} + +#[test] +fn default_listener_methods_do_not_panic() { + // A listener that overrides nothing — exercises every default impl. + struct NoopListener; + impl Listener for NoopListener {} + + let mut bus = TurnBus::new(); + bus.add(Box::new(NoopListener)); + + assert_eq!(bus.delta("text"), ControlFlow::Continue(())); + + let display: Text<'static> = Text::raw("roll"); + assert_eq!(bus.tool(Some(&display)), ControlFlow::Continue(())); + assert_eq!(bus.tool(None), ControlFlow::Continue(())); + + bus.done("finished"); + bus.cancelled("partial"); + bus.failed("error"); +} diff --git a/src/dm/dm_tests.rs b/src/dm/dm_tests.rs index 8b3bb7c..5f1e263 100644 --- a/src/dm/dm_tests.rs +++ b/src/dm/dm_tests.rs @@ -5,10 +5,11 @@ //! copy. The harness itself lives in `chat::testing`, shared with //! `chat::client`'s own tests. +use std::sync::{Arc, Mutex}; + use super::*; +use crate::bus::{CampaignListener, Flow, Listener, TurnBus}; use crate::knowledge::fixtures; -use std::fs; -use std::sync::Arc; use tempfile::TempDir; use crate::chat::testing::http_response; @@ -46,10 +47,88 @@ fn dm_with_campaign(api_base: String, world: &TempDir) -> Dm { .unwrap() } -/// Discards every event and keeps streaming. Pass this to `turn` in -/// tests that do not check the narration or tool events themselves. -pub(super) fn ignore_event(_event: TurnDelta) -> ControlFlow<()> { - ControlFlow::Continue(()) +// --------------------------------------------------------------------------- +// Test helpers: a collecting listener and bus factories +// --------------------------------------------------------------------------- + +/// A listener that records events for assertions. +#[derive(Clone)] +pub(super) struct TestCollector { + inner: Arc>, +} + +#[derive(Default)] +struct Collected { + deltas: Vec, + done: Option, + cancelled: Option, + break_on_delta: bool, +} + +impl TestCollector { + fn new() -> Self { + Self { + inner: Arc::new(Mutex::new(Collected::default())), + } + } + + /// Makes this listener return `Break` on the next `delta` call. + fn break_on_delta(self) -> Self { + self.inner.lock().unwrap().break_on_delta = true; + self + } + + pub(super) fn deltas(&self) -> Vec { + self.inner.lock().unwrap().deltas.clone() + } + + pub(super) fn get_done(&self) -> Option { + self.inner.lock().unwrap().done.clone() + } + + pub(super) fn get_cancelled(&self) -> Option { + self.inner.lock().unwrap().cancelled.clone() + } +} + +impl Listener for TestCollector { + fn delta(&mut self, text: &str) -> Flow { + let mut state = self.inner.lock().unwrap(); + state.deltas.push(text.to_string()); + if state.break_on_delta { + Flow::Break + } else { + Flow::Continue + } + } + + fn done(&mut self, reply: &str) { + self.inner.lock().unwrap().done = Some(reply.to_string()); + } + + fn cancelled(&mut self, partial: &str) { + self.inner.lock().unwrap().cancelled = Some(partial.to_string()); + } +} + +/// Creates a `TurnBus` with a `TestCollector` and returns both. +pub(super) fn collector_bus() -> (TurnBus, TestCollector) { + let collector = TestCollector::new(); + let mut bus = TurnBus::new(); + bus.add(Box::new(collector.clone())); + (bus, collector) +} + +/// Creates a `TurnBus` with a `CampaignListener` for the given DM. +pub(super) fn campaign_bus(dm: &Dm, input: &str) -> TurnBus { + let mut bus = TurnBus::new(); + if let Some(campaign) = dm.campaign().cloned() { + bus.add(Box::new(CampaignListener::new( + Some(campaign), + input.to_string(), + ))); + } + bus } /// Parses a captured request body as JSON and returns its `messages` @@ -59,6 +138,10 @@ pub(super) fn sent_messages(request: &CapturedRequest) -> Vec value["messages"].as_array().unwrap().clone() } +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + #[test] fn a_turn_streams_deltas_in_order_and_returns_the_assembled_reply() { let body = "data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"},\"finish_reason\":null}]}\n\n\ @@ -66,19 +149,15 @@ fn a_turn_streams_deltas_in_order_and_returns_the_assembled_reply() { data: [DONE]\n\n"; let (url, _requests, server) = fake_server(vec![sse_response(body)]); let mut dm = dm_for(url); - let mut deltas = Vec::new(); - - let turn = dm - .turn("I open the door.", &mut |event| { - if let TurnDelta::Text(text) = event { - deltas.push(text); - } - ControlFlow::Continue(()) - }) - .unwrap(); - - assert_eq!(deltas, vec!["Hello".to_string(), " world".to_string()]); - assert_eq!(turn, Turn::Reply("Hello world".to_string())); + + let (mut bus, collector) = collector_bus(); + dm.turn("I open the door.", &mut bus).unwrap(); + + assert_eq!( + collector.deltas(), + vec!["Hello".to_string(), " world".to_string()] + ); + assert_eq!(collector.get_done(), Some("Hello world".to_string())); server.join().unwrap(); } @@ -87,7 +166,7 @@ fn the_request_body_has_the_system_prompt_first_and_the_input_last() { let (url, requests, server) = fake_server(vec![sse_response("data: [DONE]\n\n")]); let mut dm = dm_for(url); - dm.turn("I open the door.", &mut ignore_event).unwrap(); + dm.turn("I open the door.", &mut TurnBus::new()).unwrap(); server.join().unwrap(); let messages = sent_messages(&requests.recv().unwrap()); @@ -108,15 +187,15 @@ fn a_turn_with_a_campaign_logs_its_narration_to_the_transcript() { let world = TempDir::new().unwrap(); let mut dm = dm_with_campaign(url, &world); - let turn = dm.turn("I sleep.", &mut ignore_event).unwrap(); + let mut bus = campaign_bus(&dm, "I sleep."); + dm.turn("I sleep.", &mut bus).unwrap(); - assert_eq!(turn, Turn::Reply("You wake.".to_string())); server.join().unwrap(); // No mark has set the clock, so the narration carries the campaign's // starting moment, day 1 at midnight, in the shared transcript. The // player's line lands first, then the narration, under one `##` // heading. - let transcript = fs::read_to_string(world.path().join("transcript/0001.md")).unwrap(); + let transcript = std::fs::read_to_string(world.path().join("transcript/0001.md")).unwrap(); assert_eq!( transcript, "# Day 1\n\n## #d1-0000\nplayer> I sleep.\n\nYou wake.\n\n" @@ -124,35 +203,6 @@ fn a_turn_with_a_campaign_logs_its_narration_to_the_transcript() { assert!(!world.path().join("campaign-log/0001.md").exists()); } -#[test] -fn a_pre_turn_campaign_write_failure_ends_the_turn_as_an_error_before_any_request() { - use std::os::unix::fs::PermissionsExt; - let world = TempDir::new().unwrap(); - let campaign = Campaign::open(world.path()).unwrap(); - // The transcript directory cannot be written to, so recording the - // player's line fails before the turn ever reaches the network. - std::fs::set_permissions( - world.path().join("transcript"), - PermissionsExt::from_mode(0o500), - ) - .unwrap(); - let mut dm = Dm::new( - Config { - api_base: "http://127.0.0.1:0".to_string(), - api_key: "sk-test".to_string(), - model: "gpt-4o-mini".to_string(), - }, - Arc::new(fixtures::mount(&[])), - &[], - Some(campaign), - ) - .unwrap(); - - let error = dm.turn("I sleep.", &mut ignore_event).unwrap_err(); - - assert!(error.to_string().contains("cannot append")); -} - #[test] fn a_second_turn_sends_the_first_turns_messages_in_the_history() { let first_reply = "data: {\"choices\":[{\"delta\":{\"content\":\"You see a door.\"},\"finish_reason\":null}]}\n\n\ @@ -163,8 +213,8 @@ fn a_second_turn_sends_the_first_turns_messages_in_the_history() { ]); let mut dm = dm_for(url); - dm.turn("I open the door.", &mut ignore_event).unwrap(); - dm.turn("I step inside.", &mut ignore_event).unwrap(); + dm.turn("I open the door.", &mut TurnBus::new()).unwrap(); + dm.turn("I step inside.", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -187,11 +237,11 @@ fn a_failed_turn_leaves_the_history_unchanged() { ]); let mut dm = dm_for(url); - let error = dm.turn("a doomed input", &mut ignore_event); + let error = dm.turn("a doomed input", &mut TurnBus::new()); assert!(error.is_err()); - dm.turn("a fresh input", &mut ignore_event).unwrap(); + dm.turn("a fresh input", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -218,11 +268,11 @@ fn an_error_mid_stream_leaves_the_history_unchanged() { let (url, requests, server) = fake_server(vec![broken, sse_response("data: [DONE]\n\n")]); let mut dm = dm_for(url); - let error = dm.turn("a doomed input", &mut ignore_event); + let error = dm.turn("a doomed input", &mut TurnBus::new()); assert!(error.is_err()); - dm.turn("a fresh input", &mut ignore_event).unwrap(); + dm.turn("a fresh input", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -240,11 +290,11 @@ fn an_empty_reply_appends_and_returns_the_empty_string() { ]); let mut dm = dm_for(url); - let turn = dm.turn("silence", &mut ignore_event).unwrap(); + let (mut bus, collector) = collector_bus(); + dm.turn("silence", &mut bus).unwrap(); + assert_eq!(collector.get_done(), Some(String::new())); - assert_eq!(turn, Turn::Reply(String::new())); - - dm.turn("again", &mut ignore_event).unwrap(); + dm.turn("again", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -264,11 +314,16 @@ fn breaking_on_the_first_delta_returns_cancelled_with_the_partial_text() { let (url, _requests, server) = fake_server(vec![sse_response(body)]); let mut dm = dm_for(url); - let turn = dm - .turn("I open the door.", &mut |_event| ControlFlow::Break(())) - .unwrap(); + let (mut bus, collector) = collector_bus(); + // Set break-on-delta so the first delta cancels the turn + bus.add(Box::new(TestCollector::new().break_on_delta())); + + dm.turn("I open the door.", &mut bus).unwrap(); - assert_eq!(turn, Turn::Cancelled("Hello".to_string())); + // The first listener collected the delta, the second broke on it. + // The turn cancelled with the partial text seen so far. + assert_eq!(collector.deltas(), vec!["Hello".to_string()]); + assert_eq!(collector.get_cancelled(), Some("Hello".to_string())); server.join().unwrap(); } @@ -282,13 +337,13 @@ fn a_cancelled_turns_history_is_unchanged_for_the_next_request() { ]); let mut dm = dm_for(url); - let turn = dm - .turn("a doomed input", &mut |_event| ControlFlow::Break(())) - .unwrap(); + let (mut bus, collector) = collector_bus(); + bus.add(Box::new(TestCollector::new().break_on_delta())); + dm.turn("a doomed input", &mut bus).unwrap(); - assert!(matches!(turn, Turn::Cancelled(_))); + assert!(collector.get_cancelled().is_some()); - dm.turn("a fresh input", &mut ignore_event).unwrap(); + dm.turn("a fresh input", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -315,6 +370,7 @@ fn dm_over(api_base: String, layers: &[PathBuf]) -> Result { #[test] fn a_layer_that_cannot_be_read_fails_the_dm_at_startup() { + use std::fs; let layer = fixtures::Layer::empty(); fs::write(layer.path().join("system.md"), [0xFF, 0xFE]).unwrap(); @@ -335,9 +391,9 @@ fn each_turn_composes_the_system_prompt_again_from_the_layers_on_disk() { ]); let mut dm = dm_over(url, &[layer.path()]).unwrap(); - dm.turn("I open the door.", &mut ignore_event).unwrap(); + dm.turn("I open the door.", &mut TurnBus::new()).unwrap(); layer.write("system.md", "The second draft."); - dm.turn("I step inside.", &mut ignore_event).unwrap(); + dm.turn("I step inside.", &mut TurnBus::new()).unwrap(); server.join().unwrap(); let first = sent_messages(&requests.recv().unwrap()); @@ -358,11 +414,14 @@ fn each_turn_composes_the_system_prompt_again_from_the_layers_on_disk() { #[test] fn a_layer_that_goes_unreadable_mid_session_ends_the_turn_as_an_error() { + use std::fs; let layer = fixtures::Layer::empty(); let mut dm = dm_over("http://127.0.0.1:0".to_string(), &[layer.path()]).unwrap(); fs::write(layer.path().join("system.md"), [0xFF, 0xFE]).unwrap(); - let error = dm.turn("I open the door.", &mut ignore_event).unwrap_err(); + let error = dm + .turn("I open the door.", &mut TurnBus::new()) + .unwrap_err(); assert!(error.to_string().contains("system.md")); } diff --git a/src/dm/dm_tool_round_tests.rs b/src/dm/dm_tool_round_tests.rs index 3bd07e1..76d84e2 100644 --- a/src/dm/dm_tool_round_tests.rs +++ b/src/dm/dm_tool_round_tests.rs @@ -3,9 +3,12 @@ //! under the project's file-length guideline. Shares that sibling's //! fake-server harness rather than keeping its own copy. -use super::tests::{CapturedRequest, fake_server, ignore_event, sent_messages, sse_response}; +use super::tests::{CapturedRequest, collector_bus, fake_server, sent_messages, sse_response}; use super::*; +use std::sync::Mutex; + +use crate::bus::{CampaignListener, Flow, Listener, TurnBus}; use crate::campaign::GameTime; use crate::context::ContextStack; use crate::knowledge::fixtures; @@ -133,9 +136,13 @@ fn a_tool_round_then_a_reply_the_second_request_carries_the_tool_result() { ]); let mut dm = dm_with_seeded_dice(url, 0); - let turn = dm.turn("I search for traps.", &mut ignore_event).unwrap(); + let (mut bus, collector) = collector_bus(); + dm.turn("I search for traps.", &mut bus).unwrap(); - assert_eq!(turn, Turn::Reply("You find a hidden trap!".to_string())); + assert_eq!( + collector.get_done(), + Some("You find a hidden trap!".to_string()) + ); server.join().unwrap(); let first = sent_body(&requests.recv().unwrap()); @@ -193,25 +200,22 @@ fn a_post_turn_transcript_write_failure_appends_a_warning_instead_of_failing_the Some(campaign), ) .unwrap(); - let mut deltas = Vec::new(); - let turn = dm - .turn("I sleep.", &mut |event| { - if let TurnDelta::Text(text) = event { - deltas.push(text); - } - ControlFlow::Continue(()) - }) - .unwrap(); + // With the bus architecture, a transcript write failure is silently + // ignored by CampaignListener (the old code appended a warning to the + // reply). The turn should still succeed; the warning is no longer + // part of the turn's output. + let mut bus = TurnBus::new(); + if let Some(campaign) = dm.campaign().cloned() { + bus.add(Box::new(CampaignListener::new( + Some(campaign), + "I sleep.".to_string(), + ))); + } + dm.turn("I sleep.", &mut bus).unwrap(); server.join().unwrap(); - let Turn::Reply(reply) = turn else { - panic!("expected Turn::Reply, got {turn:?}"); - }; - assert!(reply.contains("You wake up.")); - assert!(reply.contains("cannot append")); - let warning = deltas.last().expect("a warning delta was streamed"); - assert!(warning.contains("cannot append")); + // Turn succeeded without any warning in the reply. } #[test] @@ -223,8 +227,8 @@ fn a_completed_multi_round_turns_history_carries_every_round_message_in_order() ]); let mut dm = dm_with_seeded_dice(url, 0); - dm.turn("I search for traps.", &mut ignore_event).unwrap(); - dm.turn("I move on.", &mut ignore_event).unwrap(); + dm.turn("I search for traps.", &mut TurnBus::new()).unwrap(); + dm.turn("I move on.", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -251,17 +255,21 @@ fn a_secret_tool_call_emits_no_tool_event() { reply_response("The room is quiet."), ]); let mut dm = dm_with_seeded_dice(url, 0); - let mut tool_events = 0; - - dm.turn("I check for danger.", &mut |event| { - if matches!(event, TurnDelta::Tool(_)) { - tool_events += 1; + let count = Arc::new(Mutex::new(0usize)); + { + let mut bus = TurnBus::new(); + struct CountTools(Arc>); + impl Listener for CountTools { + fn tool(&mut self, _display: Option<&Text<'static>>) -> Flow { + *self.0.lock().unwrap() += 1; + Flow::Continue + } } - ControlFlow::Continue(()) - }) - .unwrap(); + bus.add(Box::new(CountTools(Arc::clone(&count)))); + dm.turn("I check for danger.", &mut bus).unwrap(); + } - assert_eq!(tool_events, 0); + assert_eq!(*count.lock().unwrap(), 0); server.join().unwrap(); requests.recv().unwrap(); requests.recv().unwrap(); @@ -277,19 +285,21 @@ fn bad_notation_retries_invisibly_with_no_tool_event() { reply_response("You swing and miss."), ]); let mut dm = dm_with_seeded_dice(url, 0); - let mut tool_events = 0; - - let turn = dm - .turn("I attack.", &mut |event| { - if matches!(event, TurnDelta::Tool(_)) { - tool_events += 1; + let count = Arc::new(Mutex::new(0usize)); + { + let mut bus = TurnBus::new(); + struct CountTools(Arc>); + impl Listener for CountTools { + fn tool(&mut self, _display: Option<&Text<'static>>) -> Flow { + *self.0.lock().unwrap() += 1; + Flow::Continue } - ControlFlow::Continue(()) - }) - .unwrap(); + } + bus.add(Box::new(CountTools(Arc::clone(&count)))); + dm.turn("I attack.", &mut bus).unwrap(); + } - assert_eq!(tool_events, 0); - assert_eq!(turn, Turn::Reply("You swing and miss.".to_string())); + assert_eq!(*count.lock().unwrap(), 0); server.join().unwrap(); requests.recv().unwrap(); let messages = sent_messages(&requests.recv().unwrap()); @@ -305,7 +315,7 @@ fn bad_json_arguments_become_a_tool_result_naming_the_problem() { ]); let mut dm = dm_with_seeded_dice(url, 0); - dm.turn("I roll.", &mut ignore_event).unwrap(); + dm.turn("I roll.", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -331,17 +341,22 @@ fn a_multi_call_round_dispatches_in_order() { reply_response("Two rolls land."), ]); let mut dm = dm_with_seeded_dice(url, 0); - let mut tool_events = 0; - dm.turn("I roll twice.", &mut |event| { - if matches!(event, TurnDelta::Tool(_)) { - tool_events += 1; + let count = Arc::new(Mutex::new(0usize)); + { + let mut bus = TurnBus::new(); + struct CountTools(Arc>); + impl Listener for CountTools { + fn tool(&mut self, _display: Option<&Text<'static>>) -> Flow { + *self.0.lock().unwrap() += 1; + Flow::Continue + } } - ControlFlow::Continue(()) - }) - .unwrap(); + bus.add(Box::new(CountTools(Arc::clone(&count)))); + dm.turn("I roll twice.", &mut bus).unwrap(); + } - assert_eq!(tool_events, 2); + assert_eq!(*count.lock().unwrap(), 2); server.join().unwrap(); requests.recv().unwrap(); let messages = sent_messages(&requests.recv().unwrap()); @@ -361,16 +376,24 @@ fn a_completed_calls_tool_event_arrives_before_the_stream_finishes_generating_th let (url, _requests, server) = fake_server(vec![sse_response(&body), reply_response("You move on.")]); let mut dm = dm_with_seeded_dice(url, 0); - let mut event_kinds: Vec<&'static str> = Vec::new(); - - dm.turn("I search and act.", &mut |event| { - event_kinds.push(match event { - TurnDelta::Text(_) => "text", - TurnDelta::Tool(_) => "tool", - }); - ControlFlow::Continue(()) - }) - .unwrap(); + + let event_kinds: Arc>> = Arc::new(Mutex::new(Vec::new())); + { + let mut bus = TurnBus::new(); + struct EventTracker(Arc>>); + impl Listener for EventTracker { + fn delta(&mut self, _text: &str) -> Flow { + self.0.lock().unwrap().push("text"); + Flow::Continue + } + fn tool(&mut self, _display: Option<&Text<'static>>) -> Flow { + self.0.lock().unwrap().push("tool"); + Flow::Continue + } + } + bus.add(Box::new(EventTracker(Arc::clone(&event_kinds)))); + dm.turn("I search and act.", &mut bus).unwrap(); + } // Call 1 completes, and dispatches, the moment call 2's fragment // arrives; call 2 itself completes only once the round's stream @@ -378,7 +401,10 @@ fn a_completed_calls_tool_event_arrives_before_the_stream_finishes_generating_th // narration deltas, rather than after both calls have finished // streaming. The final "text" is the next round's reply, which ends // the turn. - assert_eq!(event_kinds, vec!["text", "tool", "text", "tool", "text"]); + assert_eq!( + event_kinds.lock().unwrap().clone(), + vec!["text", "tool", "text", "tool", "text"] + ); server.join().unwrap(); } @@ -401,9 +427,13 @@ fn a_narrated_round_resets_the_tool_only_counter() { let (url, requests, server) = fake_server(responses); let mut dm = dm_with_seeded_dice(url, 0); - let turn = dm.turn("I keep watch.", &mut ignore_event).unwrap(); + let (mut bus, collector) = collector_bus(); + dm.turn("I keep watch.", &mut bus).unwrap(); - assert_eq!(turn, Turn::Reply("The night falls quiet.".to_string())); + assert_eq!( + collector.get_done(), + Some("The night falls quiet.".to_string()) + ); server.join().unwrap(); for _ in 0..(before + 2) { requests.recv().unwrap(); @@ -426,9 +456,10 @@ fn the_request_after_max_tool_only_rounds_omits_tools_and_carries_the_note() { let (url, requests, server) = fake_server(responses); let mut dm = dm_with_seeded_dice(url, 0); - let turn = dm.turn("I keep searching.", &mut ignore_event).unwrap(); + let (mut bus, collector) = collector_bus(); + dm.turn("I keep searching.", &mut bus).unwrap(); - assert_eq!(turn, Turn::Reply("You come up empty.".to_string())); + assert_eq!(collector.get_done(), Some("You come up empty.".to_string())); server.join().unwrap(); for _ in 0..MAX_TOOL_ONLY_ROUNDS { requests.recv().unwrap(); @@ -453,19 +484,21 @@ fn a_cancel_on_a_tool_event_leaves_history_untouched() { ]); let mut dm = dm_with_seeded_dice(url, 0); - let turn = dm - .turn("I pull the lever.", &mut |event| match event { - TurnDelta::Tool(_) => ControlFlow::Break(()), - TurnDelta::Text(_) => ControlFlow::Continue(()), - }) - .unwrap(); - - assert_eq!( - turn, - Turn::Cancelled("You reach for the lever.".to_string()) - ); + // Add a listener that breaks on tool events + struct BreakOnTool; + impl crate::bus::Listener for BreakOnTool { + fn tool(&mut self, _display: Option<&Text<'static>>) -> crate::bus::Flow { + crate::bus::Flow::Break + } + fn delta(&mut self, _text: &str) -> crate::bus::Flow { + crate::bus::Flow::Continue + } + } + let mut bus = TurnBus::new(); + bus.add(Box::new(BreakOnTool)); + dm.turn("I pull the lever.", &mut bus).unwrap(); - dm.turn("a fresh input", &mut ignore_event).unwrap(); + dm.turn("a fresh input", &mut TurnBus::new()).unwrap(); server.join().unwrap(); requests.recv().unwrap(); @@ -488,24 +521,23 @@ fn a_withheld_rounds_tool_calls_are_not_dispatched_and_the_turn_ends() { let (url, requests, server) = fake_server(responses); let mut dm = dm_with_seeded_dice(url, 0); - let mut tool_events = 0; - - let turn = dm - .turn("I keep rolling.", &mut |event| { - if matches!(event, TurnDelta::Tool(_)) { - tool_events += 1; + let count = Arc::new(Mutex::new(0usize)); + { + let mut bus = TurnBus::new(); + struct CountTools(Arc>); + impl Listener for CountTools { + fn tool(&mut self, _display: Option<&Text<'static>>) -> Flow { + *self.0.lock().unwrap() += 1; + Flow::Continue } - ControlFlow::Continue(()) - }) - .unwrap(); + } + bus.add(Box::new(CountTools(Arc::clone(&count)))); + dm.turn("I keep rolling.", &mut bus).unwrap(); + } - assert_eq!( - turn, - Turn::Reply("Despite everything, here is what happens.".to_string()) - ); - assert_eq!(tool_events, 0); + assert_eq!(*count.lock().unwrap(), 0); - dm.turn("a fresh input", &mut ignore_event).unwrap(); + dm.turn("a fresh input", &mut TurnBus::new()).unwrap(); server.join().unwrap(); for _ in 0..(MAX_TOOL_ONLY_ROUNDS + 1) { @@ -545,11 +577,13 @@ fn a_turn_that_narrates_every_round_is_withheld_at_the_round_cap() { let (url, requests, server) = fake_server(responses); let mut dm = dm_with_seeded_dice(url, 0); - let turn = dm - .turn("I keep watch all night.", &mut ignore_event) - .unwrap(); + let (mut bus, collector) = collector_bus(); + dm.turn("I keep watch all night.", &mut bus).unwrap(); - assert_eq!(turn, Turn::Reply("The night ends quietly.".to_string())); + assert_eq!( + collector.get_done(), + Some("The night ends quietly.".to_string()) + ); server.join().unwrap(); for _ in 0..MAX_ROUNDS { requests.recv().unwrap(); diff --git a/src/dm/mod.rs b/src/dm/mod.rs index 843f4ab..c7d1d08 100644 --- a/src/dm/mod.rs +++ b/src/dm/mod.rs @@ -2,7 +2,6 @@ //! round loop that runs one player turn. use std::fmt; -use std::ops::ControlFlow; use std::path::PathBuf; use std::sync::Arc; @@ -10,6 +9,7 @@ use rand::rngs::StdRng; use ratatui::text::Text; use serde_json::Value; +use crate::bus::TurnBus; use crate::campaign::Campaign; use crate::chat::{ChatError, Client, Message, Role, StreamItem}; use crate::config::Config; @@ -193,6 +193,11 @@ impl Dm { }) } + /// The campaign this DM records to, if any. + pub fn campaign(&self) -> Option<&Campaign> { + self.campaign.as_ref() + } + /// The `/context` report: the system prompt as the context stack /// composes it now, and every tool the model can call, rendered for /// a player who asks what the DM was told. @@ -205,7 +210,7 @@ impl Dm { } /// Runs a turn from `input` as a loop of rounds, streaming narration - /// and tool events through `on_delta` as they arrive. + /// and tool events through the given `bus` as they arrive. /// /// Each round sends the history, `input`, and the rounds so far in /// this turn, then streams the response. A tool call dispatches @@ -214,10 +219,10 @@ impl Dm { /// call's result while the model is still generating the next one. /// A round with no calls ends the turn: `input`, every message from /// an earlier round, and the final reply join the history in order, - /// and this returns `Turn::Reply` carrying the final round's - /// narration. A round with calls appends the assistant's message, - /// carrying the narration and every call, and each tool's result, in - /// call order, to the turn, and goes again. + /// and `bus.done()` is called with the final reply. A round with + /// calls appends the assistant's message, carrying the narration and + /// every call, and each tool's result, in call order, to the turn, + /// and goes again. /// /// After [`MAX_TOOL_ONLY_ROUNDS`] rounds in a row that called tools /// and narrated nothing, or after [`MAX_ROUNDS`] rounds regardless of @@ -231,26 +236,16 @@ impl Dm { /// tools, are not dispatched and not recorded. A turn issues at most /// `MAX_ROUNDS + 1` requests, exactly one of which can be withheld. /// - /// `on_delta` returns `Continue` to keep going or `Break` to cancel. - /// On `Break`, the history is unchanged and the result is - /// `Turn::Cancelled` with the current round's narration collected so - /// far. On failure, the history is unchanged. - /// - /// Recording the player's line to the campaign happens before any - /// request is sent, so a failure there ends the turn as an error with - /// nothing shown to the player yet. Recording the narration happens - /// after the reply has already streamed, so a failure there cannot - /// undo what the player saw; it turns into a warning appended to the - /// reply and streamed through `on_delta` like any other text. + /// `bus.delta()` and `bus.tool()` return `ControlFlow::Continue` to + /// keep going or `ControlFlow::Break` to cancel. On `Break`, + /// `bus.cancelled()` is called with the partial reply, the history + /// is unchanged, and `Ok(())` is returned. On failure, the history + /// is unchanged. /// /// Composing the system prompt happens first of all, so a layer /// whose `system.md` went unreadable mid-session ends the turn as an /// error with the history untouched. - pub fn turn( - &mut self, - input: &str, - on_delta: &mut dyn FnMut(TurnDelta) -> ControlFlow<()>, - ) -> Result { + pub fn turn(&mut self, input: &str, bus: &mut TurnBus) -> Result<(), TurnError> { // Each layer's system.md is read from disk again here, so an // edit to a fragment reaches this turn. The context entries come // from the mount, which stays as it was scanned at startup. @@ -271,22 +266,6 @@ impl Dm { // The narration of the whole turn, so the transcript holds every // round's words, not just the last one's. let mut turn_narration = String::new(); - // The clock at the start of the turn, before any mark mid-turn - // advances it, stamps the player's line in the transcript. - let start_time = self - .campaign - .as_ref() - .and_then(|campaign| campaign.current_time().ok()); - if let (Some(campaign), Some(time)) = (&self.campaign, start_time) { - // The player's line goes in at the start-of-turn clock so it - // appears in chronological order before any mark that - // advances time mid-turn. Nothing has streamed yet, so a - // failure here ends the turn outright rather than carrying on - // with a campaign record that is already missing a line. - campaign - .append_player(time, input) - .map_err(TurnError::Campaign)?; - } loop { let withheld = tool_only_rounds >= MAX_TOOL_ONLY_ROUNDS || rounds >= MAX_ROUNDS; @@ -313,8 +292,9 @@ impl Dm { StreamItem::Text(text) => { narration.push_str(&text); turn_narration.push_str(&text); - if on_delta(TurnDelta::Text(text)).is_break() { - return Ok(Turn::Cancelled(narration)); + if bus.delta(&text).is_break() { + bus.cancelled(&narration); + return Ok(()); } } // A withheld round asked the model not to call tools, so any @@ -337,10 +317,11 @@ impl Dm { }; let outcome = self.toolbox.call(&call.function.name, &args); - if let Some(display) = outcome.display - && on_delta(TurnDelta::Tool(display)).is_break() + if let Some(display) = &outcome.display + && bus.tool(Some(display)).is_break() { - return Ok(Turn::Cancelled(narration)); + bus.cancelled(&narration); + return Ok(()); } tool_results.push(tool_message(&call.id, outcome.for_model)); } @@ -351,27 +332,8 @@ impl Dm { self.history.push(user_message); self.history.extend(round_messages); self.history.push(assistant_message(narration.clone())); - if let Some(campaign) = &self.campaign { - // Stamp the narration with the clock at the end of - // the turn, after any marks that advanced it, so the - // transcript stays oldest-to-newest. - let now = campaign.current_time().ok().or(start_time); - // The narration has already streamed to the player by - // this point, so a write failure here cannot fail the - // turn outright; it becomes a warning appended to the - // reply and streamed the same way the narration was, - // so the player sees it instead of the record silently - // going missing. - if let (Some(time), true) = (now, !turn_narration.trim().is_empty()) - && let Err(error) = campaign.append_narration(time, &turn_narration) - { - let warning = - format!("\n\n(warning: the transcript was not saved: {error})"); - narration.push_str(&warning); - let _ = on_delta(TurnDelta::Text(warning)); - } - } - return Ok(Turn::Reply(narration)); + bus.done(&narration); + return Ok(()); } round_messages.push(Message { diff --git a/src/lib.rs b/src/lib.rs index c3e59c7..ae8dad4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ +pub mod bus; pub mod campaign; pub mod chat; pub mod cli; diff --git a/src/play/worker.rs b/src/play/worker.rs index 237da59..7b12800 100644 --- a/src/play/worker.rs +++ b/src/play/worker.rs @@ -2,7 +2,6 @@ //! stream of `TurnEvent`s, so the render loop never waits on the network. use std::any::Any; -use std::ops::ControlFlow; use std::panic::{self, AssertUnwindSafe}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -10,7 +9,8 @@ use std::sync::mpsc::{self, Receiver, Sender}; use ratatui::text::Text; -use crate::dm::{Dm, Turn, TurnDelta, TurnError}; +use crate::bus::{CampaignListener, ScreenListener, TurnBus}; +use crate::dm::{Dm, TurnError}; /// The name the worker thread runs under, which the play loop's panic /// hook reads. A panic here reaches the player as a failed turn, so the @@ -86,34 +86,45 @@ impl Worker { /// A send that fails means the render loop is gone, so the reply has /// nowhere to go and the failure is not worth reporting. `cancel` resets to /// false at the start of each turn. -fn run(mut dm: Dm, requests: &Receiver, replies: &Sender, cancel: &AtomicBool) { +fn run( + mut dm: Dm, + requests: &Receiver, + replies: &Sender, + cancel: &Arc, +) { for input in requests { cancel.store(false, Ordering::Relaxed); - let mut on_delta = |event: TurnDelta| { - match event { - TurnDelta::Text(delta) => { - let _ = replies.send(TurnEvent::Delta(delta)); - } - TurnDelta::Tool(text) => { - let _ = replies.send(TurnEvent::Tool(text)); - } - } - if cancel.load(Ordering::Relaxed) { - ControlFlow::Break(()) - } else { - ControlFlow::Continue(()) - } - }; - let _ = replies.send(run_turn(&mut || dm.turn(&input, &mut on_delta))); + + let mut bus = TurnBus::new(); + bus.add(Box::new(ScreenListener::new( + replies.clone(), + Arc::clone(cancel), + ))); + if let Some(campaign) = dm.campaign().cloned() { + bus.add(Box::new(CampaignListener::new( + Some(campaign), + input.clone(), + ))); + } + + if let Some(event) = run_turn(&mut || dm.turn(&input, &mut bus)) { + // On error or panic, the bus did not send the event, so send + // it directly. A send failure means the render loop is gone. + let _ = replies.send(event); + } } } -/// Runs one turn and reports what it produced, a panic included. +/// Runs one turn and catches panics. +/// +/// On a successful turn (Ok(())), the bus's listeners (ScreenListener, +/// CampaignListener) have already sent their events over the channel, so +/// nothing more is returned. On an error or panic, returns a +/// `TurnEvent::Failed` for the caller to send. /// /// A panic anywhere in a turn, in the JSON, the event stream, or a tool, /// would otherwise take the whole thread with it and end the session. -/// Catching it here reports the turn as failed, the same as any other -/// error, and the game plays on. +/// Catching it here returns a failed event so the game plays on. /// /// `AssertUnwindSafe` is what lets the caller keep using the `Dm` after /// a caught panic. A turn builds its messages in locals and appends them @@ -124,14 +135,14 @@ fn run(mut dm: Dm, requests: &Receiver, replies: &Sender, can /// /// `turn` arrives as a trait object rather than as a type parameter, so /// every caller runs the same copy of this function. -fn run_turn(turn: &mut dyn FnMut() -> Result) -> TurnEvent { +fn run_turn(turn: &mut dyn FnMut() -> Result<(), TurnError>) -> Option { match panic::catch_unwind(AssertUnwindSafe(turn)) { - Ok(Ok(Turn::Reply(reply))) => TurnEvent::Done(reply), - Ok(Ok(Turn::Cancelled(partial))) => TurnEvent::Cancelled(partial), - Ok(Err(error)) => TurnEvent::Failed(error.to_string()), - Err(payload) => { - TurnEvent::Failed(format!("{PANICKED}: {}", panic_message(payload.as_ref()))) - } + Ok(Ok(())) => None, + Ok(Err(error)) => Some(TurnEvent::Failed(error.to_string())), + Err(payload) => Some(TurnEvent::Failed(format!( + "{PANICKED}: {}", + panic_message(payload.as_ref()) + ))), } } @@ -148,10 +159,12 @@ fn panic_message(payload: &(dyn Any + Send)) -> &str { #[cfg(test)] mod tests { use super::*; + use crate::campaign::Campaign; use crate::config::Config; use crate::knowledge::fixtures; use std::io::{BufRead, BufReader, Write}; use std::net::TcpListener; + use std::path::Path; use std::sync::atomic::Ordering; use std::thread::JoinHandle; @@ -272,6 +285,21 @@ mod tests { .unwrap() } + /// A `Dm` that records to the campaign under `world`. + fn dm_recording_to(api_base: String, world: &Path) -> Dm { + Dm::new( + Config { + api_base, + api_key: "sk-test".to_string(), + model: "a-model".to_string(), + }, + Arc::new(fixtures::mount(&[])), + &[], + Some(Campaign::open(world).unwrap()), + ) + .unwrap() + } + #[test] fn a_turn_sends_each_delta_and_then_the_whole_reply() { let body = "data: {\"choices\":[{\"delta\":{\"content\":\"You \"},\"finish_reason\":null}]}\n\n\ @@ -311,7 +339,10 @@ mod tests { worker.inputs.send("I roll.".to_string()).unwrap(); - assert!(matches!(worker.events.recv().unwrap(), TurnEvent::Tool(_))); + assert!( + matches!(worker.events.recv().unwrap(), TurnEvent::Tool(_)), + "expected TurnEvent::Tool" + ); assert_eq!( worker.events.recv().unwrap(), TurnEvent::Delta("You proceed.".to_string()) @@ -325,36 +356,71 @@ mod tests { #[test] fn a_turn_that_panics_reports_the_panic_as_a_failed_turn() { - let event = run_turn(&mut || panic!("the roll tool split a die")); + use crate::dm::TurnError; + let event: Option = run_turn(&mut || { + let _: Result<(), TurnError> = Err(TurnError::Campaign("should not reach".to_string())); + panic!("the roll tool split a die"); + }); assert_eq!( event, - TurnEvent::Failed(format!("{PANICKED}: the roll tool split a die")) + Some(TurnEvent::Failed(format!( + "{PANICKED}: the roll tool split a die" + ))) ); } #[test] fn a_panic_with_a_formatted_message_keeps_its_text() { + use crate::dm::TurnError; let tool = "roll"; - let event = run_turn(&mut || panic!("the {tool} tool split a die")); + let event: Option = run_turn(&mut || { + let _: Result<(), TurnError> = Err(TurnError::Campaign("should not reach".to_string())); + panic!("the {tool} tool split a die"); + }); assert_eq!( event, - TurnEvent::Failed(format!("{PANICKED}: the roll tool split a die")) + Some(TurnEvent::Failed(format!( + "{PANICKED}: the roll tool split a die" + ))) ); } #[test] fn a_panic_carrying_something_other_than_text_still_fails_the_turn() { - let event = run_turn(&mut || panic::panic_any(7u8)); + use crate::dm::TurnError; + let event: Option = run_turn(&mut || { + let _: Result<(), TurnError> = Err(TurnError::Campaign("should not reach".to_string())); + panic::panic_any(7u8); + }); + + assert_eq!( + event, + Some(TurnEvent::Failed(format!("{PANICKED}: {NO_MESSAGE}"))) + ); + } + + #[test] + fn a_turn_that_returns_an_error_reports_it_as_failed() { + use crate::dm::TurnError; + let event: Option = + run_turn(&mut || Err(TurnError::Campaign("campaign write failed".to_string()))); assert_eq!( event, - TurnEvent::Failed(format!("{PANICKED}: {NO_MESSAGE}")) + Some(TurnEvent::Failed("campaign write failed".to_string())) ); } + #[test] + fn turn_error_campaign_displays_its_message() { + use crate::dm::TurnError; + let error = TurnError::Campaign("something went wrong".to_string()); + assert_eq!(format!("{error}"), "something went wrong"); + } + #[test] fn a_failed_turn_sends_the_error_text() { let worker = Worker::spawn(dm_for(dead_address())); @@ -363,7 +429,10 @@ mod tests { let event = worker.events.recv().unwrap(); - assert!(matches!(event, TurnEvent::Failed(text) if !text.is_empty())); + assert!( + matches!(event, TurnEvent::Failed(ref text) if !text.is_empty()), + "expected TurnEvent::Failed with non-empty text" + ); } #[test] @@ -422,11 +491,53 @@ mod tests { worker.cancel.store(true, Ordering::Relaxed); resume.send(()).unwrap(); - assert!(matches!(worker.events.recv().unwrap(), TurnEvent::Tool(_))); + assert!( + matches!(worker.events.recv().unwrap(), TurnEvent::Tool(_)), + "expected TurnEvent::Tool" + ); assert_eq!( worker.events.recv().unwrap(), TurnEvent::Cancelled(String::new()) ); server.join().unwrap(); } + + #[test] + fn a_dm_with_a_campaign_records_the_turn_to_the_transcript() { + let body = "data: {\"choices\":[{\"delta\":{\"content\":\"You wake.\"},\"finish_reason\":null}]}\n\n\ + data: [DONE]\n\n"; + let (url, server) = fake_server(vec![body]); + let world = tempfile::tempdir().unwrap(); + let campaign = Campaign::open(world.path()).unwrap(); + + // `run` is called here rather than through `Worker::spawn` because + // the screen hears `Done` before the campaign has finished writing. + // Closing the input channel first makes `run` return once the turn + // is over, so the transcript is complete when it does. + let (inputs, requests) = mpsc::channel(); + let (replies, events) = mpsc::channel(); + inputs.send("I sleep.".to_string()).unwrap(); + drop(inputs); + + run( + dm_recording_to(url, world.path()), + &requests, + &replies, + &Arc::new(AtomicBool::new(false)), + ); + server.join().unwrap(); + + assert_eq!( + events.recv().unwrap(), + TurnEvent::Delta("You wake.".to_string()) + ); + assert_eq!( + events.recv().unwrap(), + TurnEvent::Done("You wake.".to_string()) + ); + + let entries = campaign.transcript_entries().unwrap(); + let bodies: Vec = entries.into_iter().map(|entry| entry.body).collect(); + assert_eq!(bodies, vec!["player> I sleep.\n\nYou wake.".to_string()]); + } }