diff --git a/rust/crates/sdk/src/session/mod.rs b/rust/crates/sdk/src/session/mod.rs index c7c61a58..e8923ed5 100644 --- a/rust/crates/sdk/src/session/mod.rs +++ b/rust/crates/sdk/src/session/mod.rs @@ -13,7 +13,7 @@ pub use terminal_scope::{TerminalScopeRecord, TerminalScopeStore}; use std::collections::{BTreeMap, HashMap}; use std::fmt::{Debug, Formatter}; use std::panic::{catch_unwind, AssertUnwindSafe}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, MutexGuard}; use std::thread; use std::time::{SystemTime, UNIX_EPOCH}; @@ -209,6 +209,7 @@ struct SessionEventPublisher { session_id: String, invocation_id: String, sequence: std::sync::atomic::AtomicU64, + publication_lock: Mutex<()>, sender: mpsc::UnboundedSender, } @@ -222,11 +223,23 @@ impl SessionEventPublisher { session_id, invocation_id, sequence: std::sync::atomic::AtomicU64::new(0), + publication_lock: Mutex::new(()), sender, } } - fn build(&self, event_type: SessionEventType, payload: Value, durable: bool) -> SessionEvent { + fn lock(&self) -> MutexGuard<'_, ()> { + self.publication_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + fn build_locked( + &self, + event_type: SessionEventType, + payload: Value, + durable: bool, + ) -> SessionEvent { let sequence = self .sequence .fetch_add(1, std::sync::atomic::Ordering::SeqCst) @@ -244,11 +257,34 @@ impl SessionEventPublisher { } fn publish(&self, event_type: SessionEventType, payload: Value, durable: bool) -> SessionEvent { - let event = self.build(event_type, payload, durable); + let _guard = self.lock(); + self.publish_locked(event_type, payload, durable) + } + + fn publish_locked( + &self, + event_type: SessionEventType, + payload: Value, + durable: bool, + ) -> SessionEvent { + let event = self.build_locked(event_type, payload, durable); let _ = self.sender.send(event.clone()); event } + fn publish_steer(&self, control: &TurnControlSignal, prompt: &str) -> bool { + let _guard = self.lock(); + if !control.steer(prompt.to_string()) { + return false; + } + self.publish_locked( + SessionEventType::TurnSteered, + json!({ "prompt": prompt }), + false, + ); + true + } + fn publish_runtime_event(&self, event: ConversationEvent) { let (event_type, payload) = match event { ConversationEvent::AgentBegin { session_id: _ } => { @@ -361,11 +397,38 @@ impl SessionRouter { /// Resolve the most recently recorded session for a correlation key. pub fn session_for_correlation_key(&self, correlation_key: &str) -> SdkResult> { validate_correlation_key(correlation_key)?; - Ok(self + { + let active = self + .inner + .active + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut matches = active.iter().filter(|(_, invocation)| { + invocation.correlation_key.as_deref() == Some(correlation_key) + }); + if let Some((session_id, _)) = matches.next() { + if matches.next().is_some() { + return Err(SdkError::conflict(format!( + "correlation key '{correlation_key}' has multiple active sessions" + ))); + } + return Ok(Some(session_id.clone())); + } + } + + let has_unfinished_invocation = self .inner - .terminal_scopes - .get(correlation_key)? - .map(|scope| scope.session_id)) + .reconciliation + .list_by_correlation_key(correlation_key)? + .iter() + .any(|record| record.status != InvocationStatus::Terminal); + if has_unfinished_invocation { + return Err(SdkError::unavailable(format!( + "correlation key '{correlation_key}' has an invocation awaiting reconciliation" + ))); + } + let scope = self.inner.terminal_scopes.get(correlation_key)?; + Ok(scope.map(|scope| scope.session_id)) } /// Start a prompt and return immediately with an ordered event receiver. @@ -404,6 +467,7 @@ impl SessionRouter { outcome: None, uncertainty_reason: None, terminal_event: None, + terminal_scope_revision: None, }; self.inner.reconciliation.insert(record)?; @@ -447,8 +511,9 @@ impl SessionRouter { if let Err(error) = inner.reconciliation.mark_running(&worker_invocation_id) { finish_uncertain( &inner, - &worker_session_id, &worker_invocation_id, + &worker_control, + &worker_publisher, &error.to_string(), ); remove_active(&inner, &worker_session_id, &worker_invocation_id); @@ -469,6 +534,7 @@ impl SessionRouter { &worker_session_id, &worker_invocation_id, worker_correlation_key.as_deref(), + &worker_control, &worker_publisher, run_result, ), @@ -476,8 +542,9 @@ impl SessionRouter { let reason = panic_message(panic); finish_uncertain( &inner, - &worker_session_id, &worker_invocation_id, + &worker_control, + &worker_publisher, &reason, ); } @@ -493,6 +560,7 @@ impl SessionRouter { session_id, &invocation_id, correlation_key, + &control, &publisher, SessionRunResult::Failed { session, message }, ); @@ -518,27 +586,28 @@ impl SessionRouter { validate_invocation_id(invocation_id)?; let prompt = override_prompt.into(); validate_prompt(&prompt)?; - let active = self - .inner - .active - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let invocation = active - .get(session_id) - .ok_or_else(|| SdkError::not_found("session has no active invocation"))?; - if invocation.invocation_id != invocation_id { - return Err(SdkError::not_found(format!( - "invocation `{invocation_id}` is not active for session `{session_id}`" - ))); - } - if !invocation.control.steer(prompt.clone()) { + let (control, publisher) = { + let active = self + .inner + .active + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let invocation = active + .get(session_id) + .ok_or_else(|| SdkError::not_found("session has no active invocation"))?; + if invocation.invocation_id != invocation_id { + return Err(SdkError::not_found(format!( + "invocation `{invocation_id}` is not active for session `{session_id}`" + ))); + } + ( + invocation.control.clone(), + Arc::clone(&invocation.event_publisher), + ) + }; + if !publisher.publish_steer(&control, &prompt) { return Err(SdkError::conflict("invocation no longer accepts steering")); } - invocation.event_publisher.publish( - SessionEventType::TurnSteered, - json!({ "prompt": prompt }), - false, - ); Ok(()) } @@ -554,7 +623,7 @@ impl SessionRouter { validate_prompt(&prompt)?; validate_optional_correlation_key(correlation_key)?; - let (mut done, inherited_correlation) = { + let (mut done, inherited_correlation, previous_invocation_id) = { let active = self .inner .active @@ -566,9 +635,10 @@ impl SessionRouter { ( Some(invocation.done.clone()), invocation.correlation_key.clone(), + Some(invocation.invocation_id.clone()), ) } - None => (None, None), + None => (None, None, None), } }; if let Some(receiver) = done.as_mut() { @@ -579,6 +649,19 @@ impl SessionRouter { } } + if let Some(invocation_id) = previous_invocation_id { + let record = self + .inner + .reconciliation + .get(&invocation_id)? + .ok_or_else(|| SdkError::not_found("aborted invocation record was not found"))?; + if record.status != InvocationStatus::Terminal { + return Err(SdkError::unavailable(format!( + "invocation `{invocation_id}` has no durable terminal outcome" + ))); + } + } + let effective_correlation = correlation_key .map(str::to_string) .or(inherited_correlation); @@ -795,9 +878,13 @@ fn finish_invocation( session_id: &str, invocation_id: &str, correlation_key: Option<&str>, + control: &TurnControlSignal, publisher: &SessionEventPublisher, result: SessionRunResult, ) { + let _publication_guard = publisher.lock(); + control.close(); + let (outcome, message_count, status, error_message) = match result { SessionRunResult::Completed(session) => ( InvocationOutcome::Completed { @@ -829,7 +916,19 @@ fn finish_invocation( if let Some(error) = error_message { payload["error"] = Value::String(error); } - let terminal_event = publisher.build(SessionEventType::TurnEnded, payload, true); + let terminal_scope_revision = match correlation_key { + Some(key) => match inner.terminal_scopes.set(key, session_id) { + Ok(scope) => Some(scope.revision), + Err(error) => { + let _ = inner + .reconciliation + .mark_uncertain(invocation_id, error.to_string()); + return; + } + }, + None => None, + }; + let terminal_event = publisher.build_locked(SessionEventType::TurnEnded, payload, true); let stored_event = match serde_json::to_value(&terminal_event) { Ok(event) => event, Err(error) => { @@ -839,14 +938,13 @@ fn finish_invocation( return; } }; - match inner - .reconciliation - .record_terminal(invocation_id, outcome, Some(stored_event)) - { + match inner.reconciliation.record_terminal_with_scope_revision( + invocation_id, + outcome, + Some(stored_event), + terminal_scope_revision, + ) { Ok(()) => { - if let Some(key) = correlation_key { - let _ = inner.terminal_scopes.set(key, session_id); - } let _ = publisher.sender.send(terminal_event); } Err(error) => { @@ -861,7 +959,7 @@ fn recover_terminal_scopes( reconciliation: &InvocationReconciliationStore, terminal_scopes: &TerminalScopeStore, ) -> SdkResult<()> { - let mut latest_by_key = BTreeMap::::new(); + let mut latest_by_key = BTreeMap::::new(); for record in reconciliation.list()? { if record.status != InvocationStatus::Terminal { continue; @@ -869,29 +967,39 @@ fn recover_terminal_scopes( let Some(key) = record.correlation_key.as_ref() else { continue; }; - if !latest_by_key - .get(key) - .is_some_and(|current| current.updated_at_ms >= record.updated_at_ms) - { - latest_by_key.insert(key.clone(), record); + let Some(revision) = record.terminal_scope_revision else { + continue; + }; + match latest_by_key.get(key) { + Some((current_revision, current_session)) if *current_revision == revision => { + if current_session != &record.session_id { + return Err(SdkError::internal(format!( + "terminal scope revision {revision} conflicts for key '{key}'" + ))); + } + } + Some((current_revision, _)) if *current_revision > revision => {} + _ => { + latest_by_key.insert(key.clone(), (revision, record.session_id)); + } } } - for (key, record) in latest_by_key { + for (key, (revision, session_id)) in latest_by_key { validate_correlation_key(&key)?; - let existing = terminal_scopes.get(&key)?; - if existing.is_none_or(|scope| scope.updated_at_ms < record.updated_at_ms) { - terminal_scopes.set(key, record.session_id)?; - } + terminal_scopes.restore_if_newer(&key, &session_id, revision)?; } Ok(()) } fn finish_uncertain( inner: &SessionRouterInner, - _session_id: &str, invocation_id: &str, + control: &TurnControlSignal, + publisher: &SessionEventPublisher, reason: &str, ) { + let _publication_guard = publisher.lock(); + control.close(); let _ = inner .reconciliation .mark_uncertain(invocation_id, reason.to_string()); diff --git a/rust/crates/sdk/src/session/reconciliation.rs b/rust/crates/sdk/src/session/reconciliation.rs index 0ee3cbe1..da1a74c6 100644 --- a/rust/crates/sdk/src/session/reconciliation.rs +++ b/rust/crates/sdk/src/session/reconciliation.rs @@ -53,6 +53,7 @@ pub struct InvocationRecord { pub outcome: Option, pub uncertainty_reason: Option, pub terminal_event: Option, + pub terminal_scope_revision: Option, } /// File-backed store for invocation lifecycle reconciliation. @@ -110,6 +111,17 @@ impl InvocationReconciliationStore { invocation_id: &str, outcome: InvocationOutcome, terminal_event: Option, + ) -> SdkResult<()> { + self.record_terminal_with_scope_revision(invocation_id, outcome, terminal_event, None) + } + + /// Records a terminal result together with its durable correlation-scope revision. + pub fn record_terminal_with_scope_revision( + &self, + invocation_id: &str, + outcome: InvocationOutcome, + terminal_event: Option, + terminal_scope_revision: Option, ) -> SdkResult<()> { let mut record = self.required_record(invocation_id)?; if record.status == InvocationStatus::Terminal { @@ -122,6 +134,7 @@ impl InvocationReconciliationStore { record.outcome = Some(outcome); record.uncertainty_reason = None; record.terminal_event = terminal_event; + record.terminal_scope_revision = terminal_scope_revision; record.updated_at_ms = updated_at_ms(record.updated_at_ms); self.write_record(&record) } @@ -362,6 +375,7 @@ mod tests { outcome: None, uncertainty_reason: None, terminal_event: None, + terminal_scope_revision: None, } } diff --git a/rust/crates/sdk/src/session/terminal_scope.rs b/rust/crates/sdk/src/session/terminal_scope.rs index 7162876b..8b2ca87a 100644 --- a/rust/crates/sdk/src/session/terminal_scope.rs +++ b/rust/crates/sdk/src/session/terminal_scope.rs @@ -3,6 +3,7 @@ use std::{ fs, path::PathBuf, + sync::Mutex, time::{SystemTime, UNIX_EPOCH}, }; @@ -13,12 +14,14 @@ use crate::error::{SdkError, SdkResult}; use super::reconciliation::atomic_write_json; const MAX_CORRELATION_KEY_BYTES: usize = 512; +static TERMINAL_SCOPE_WRITE_LOCK: Mutex<()> = Mutex::new(()); /// Persisted terminal-session scope for one correlation key. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TerminalScopeRecord { pub correlation_key: String, pub session_id: String, + pub revision: u64, pub updated_at_ms: u64, } @@ -40,24 +43,69 @@ impl TerminalScopeStore { &self, correlation_key: impl Into, session_id: impl Into, - ) -> SdkResult<()> { + ) -> SdkResult { let correlation_key = correlation_key.into(); validate_correlation_key(&correlation_key)?; - let path = self.record_path(&correlation_key); - let previous_updated_at = self - .get(&correlation_key)? - .map_or(0, |record| record.updated_at_ms); + let _guard = TERMINAL_SCOPE_WRITE_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let previous = self.read_record(&correlation_key)?; + let revision = previous + .as_ref() + .map_or(Some(1), |record| record.revision.checked_add(1)) + .ok_or_else(|| SdkError::internal("terminal scope revision is exhausted"))?; + let previous_updated_at = previous.map_or(0, |record| record.updated_at_ms); let record = TerminalScopeRecord { - correlation_key, + correlation_key: correlation_key.clone(), session_id: session_id.into(), + revision, updated_at_ms: current_timestamp_ms().max(previous_updated_at), }; - atomic_write_json(&path, &record) + atomic_write_json(&self.record_path(&correlation_key), &record)?; + Ok(record) + } + + /// Rebuilds a mapping from durable invocation history without moving it backward. + pub fn restore_if_newer( + &self, + correlation_key: &str, + session_id: &str, + revision: u64, + ) -> SdkResult<()> { + validate_correlation_key(correlation_key)?; + let _guard = TERMINAL_SCOPE_WRITE_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let previous = self.read_record(correlation_key)?; + match previous.as_ref() { + Some(record) if record.revision > revision => return Ok(()), + Some(record) if record.revision == revision && record.session_id == session_id => { + return Ok(()) + } + Some(record) if record.revision == revision => { + return Err(SdkError::internal(format!( + "terminal scope revision {revision} conflicts for key '{correlation_key}'" + ))); + } + _ => {} + } + let record = TerminalScopeRecord { + correlation_key: correlation_key.to_string(), + session_id: session_id.to_string(), + revision, + updated_at_ms: current_timestamp_ms() + .max(previous.map_or(0, |record| record.updated_at_ms)), + }; + atomic_write_json(&self.record_path(correlation_key), &record) } /// Loads the session associated with a correlation key, if present. pub fn get(&self, correlation_key: &str) -> SdkResult> { validate_correlation_key(correlation_key)?; + self.read_record(correlation_key) + } + + fn read_record(&self, correlation_key: &str) -> SdkResult> { let path = self.record_path(correlation_key); let bytes = match fs::read(&path) { Ok(bytes) => bytes, @@ -162,6 +210,26 @@ mod tests { assert_eq!(loaded.session_id, "session-1"); } + #[test] + fn scope_revisions_increase_and_recovery_never_moves_backward() { + let root = TestDirectory::new(); + let store = TerminalScopeStore::new(&root.0); + let first = store.set("request-1", "session-1").unwrap(); + let second = store.set("request-1", "session-2").unwrap(); + assert_eq!(second.revision, first.revision + 1); + + store + .restore_if_newer("request-1", "session-1", first.revision) + .expect("stale scope restoration should be ignored"); + assert_eq!( + store.get("request-1").unwrap().unwrap().session_id, + "session-2" + ); + assert!(store + .restore_if_newer("request-1", "other-session", second.revision) + .is_err()); + } + #[test] fn encoded_keys_are_isolated_and_cannot_traverse_paths() { let root = TestDirectory::new(); diff --git a/rust/crates/sdk/tests/session_lifecycle.rs b/rust/crates/sdk/tests/session_lifecycle.rs index ab53f583..7f368b1f 100644 --- a/rust/crates/sdk/tests/session_lifecycle.rs +++ b/rust/crates/sdk/tests/session_lifecycle.rs @@ -1,5 +1,5 @@ use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::{mpsc, Arc, Mutex}; +use std::sync::{mpsc, Arc, Barrier, Mutex}; use std::thread; use std::time::Duration; @@ -897,7 +897,7 @@ fn reconciliation_returns_not_found_for_unknown_invocation() { fn startup_recovers_running_invocations_as_uncertain() { let (_temp, store, router) = setup(); let session = router - .create_session(None, None) + .create_session(None, Some("recover-me")) .expect("session should persist"); let invocation_id = Uuid::new_v4().to_string(); let reconciliation = sdk::InvocationReconciliationStore::new( @@ -914,11 +914,16 @@ fn startup_recovers_running_invocations_as_uncertain() { outcome: None, uncertainty_reason: None, terminal_event: None, + terminal_scope_revision: None, }) .expect("running record"); let restarted = SessionRouter::new(store, Arc::new(EchoRunner)).expect("new router"); let record = restarted.reconcile(&invocation_id).unwrap(); assert_eq!(record.status, InvocationStatus::Uncertain); + let error = restarted + .session_for_correlation_key("recover-me") + .unwrap_err(); + assert_eq!(error.code, ErrorCode::Unavailable); } #[test] @@ -938,6 +943,41 @@ fn terminal_scope_survives_router_restart() { assert!(temp.path().exists()); } +#[tokio::test] +async fn terminal_scope_write_failure_keeps_outcome_uncertain_and_blocks_stale_routing() { + let (started_tx, started_rx) = mpsc::sync_channel(1); + let (release_tx, release_rx) = mpsc::sync_channel(1); + let runner = Arc::new(GateRunner { + started: Mutex::new(Some(started_tx)), + release: Mutex::new(release_rx), + }); + let (_temp, store, router) = setup_with(runner); + let stale_session = router.create_session(None, Some("repair-key")).unwrap(); + let current_session = router.create_session(None, None).unwrap(); + let invocation = router + .prompt(¤t_session.session_id, "complete", Some("repair-key")) + .unwrap(); + let invocation_id = invocation.invocation_id.clone(); + started_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + + let scope_root = store.sessions_dir().join("sdk").join("terminal-scopes"); + std::fs::remove_dir_all(&scope_root).unwrap(); + std::fs::write(&scope_root, "block the scope store").unwrap(); + release_tx.send(()).unwrap(); + + let events = collect_events(invocation).await; + assert_eq!( + events.last().unwrap().event_type, + SessionEventType::AgentEnd + ); + assert_eq!( + router.reconcile(&invocation_id).unwrap().status, + InvocationStatus::Uncertain + ); + assert!(router.session_for_correlation_key("repair-key").is_err()); + assert_ne!(stale_session.session_id, current_session.session_id); +} + #[test] fn terminal_scope_keys_do_not_collide() { let (_temp, _store, router) = setup(); @@ -958,6 +998,38 @@ fn terminal_scope_keys_do_not_collide() { ); } +#[tokio::test] +async fn correlation_lookup_rejects_multiple_active_sessions() { + let (started_tx, started_rx) = mpsc::sync_channel(1); + let (release_tx, release_rx) = mpsc::sync_channel(2); + let runner = Arc::new(GateRunner { + started: Mutex::new(Some(started_tx)), + release: Mutex::new(release_rx), + }); + let (_temp, _store, router) = setup_with(runner); + let first = router.create_session(None, Some("active-key")).unwrap(); + let second = router.create_session(None, Some("active-key")).unwrap(); + let first_invocation = router + .prompt(&first.session_id, "first", Some("active-key")) + .unwrap(); + let second_invocation = router + .prompt(&second.session_id, "second", Some("active-key")) + .unwrap(); + started_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + + assert_eq!( + router + .session_for_correlation_key("active-key") + .unwrap_err() + .code, + ErrorCode::Conflict + ); + release_tx.send(()).unwrap(); + release_tx.send(()).unwrap(); + collect_events(first_invocation).await; + collect_events(second_invocation).await; +} + #[tokio::test] async fn session_router_clones_share_active_invocations() { let (started_tx, started_rx) = mpsc::sync_channel(1); @@ -1274,6 +1346,7 @@ fn new_router_recovers_only_unfinished_invocations() { outcome: Some(InvocationOutcome::Completed { message_count: 0 }), uncertainty_reason: None, terminal_event: Some(json!({ "type": "turn_ended" })), + terminal_scope_revision: Some(1), }) .unwrap(); let restarted = SessionRouter::new(store, Arc::new(EchoRunner)).unwrap(); @@ -1289,6 +1362,43 @@ fn new_router_recovers_only_unfinished_invocations() { ); } +#[test] +fn terminal_scope_recovery_uses_revision_when_timestamps_tie() { + let (_temp, store, router) = setup(); + let first = router.create_session(None, Some("revision-tie")).unwrap(); + let latest = router.create_session(None, None).unwrap(); + let reconciliation = sdk::InvocationReconciliationStore::new( + store.sessions_dir().join("sdk").join("reconciliation"), + ); + for (invocation_id, session_id, revision) in [ + ("invocation-a", first.session_id.as_str(), 1), + ("invocation-z", latest.session_id.as_str(), 2), + ] { + reconciliation + .insert(InvocationRecord { + invocation_id: invocation_id.to_string(), + session_id: session_id.to_string(), + correlation_key: Some("revision-tie".to_string()), + status: InvocationStatus::Terminal, + accepted_at_ms: 10, + updated_at_ms: 20, + outcome: Some(InvocationOutcome::Completed { message_count: 1 }), + uncertainty_reason: None, + terminal_event: Some(json!({ "type": "turn_ended" })), + terminal_scope_revision: Some(revision), + }) + .unwrap(); + } + + let restarted = SessionRouter::new(store, Arc::new(EchoRunner)).unwrap(); + assert_eq!( + restarted + .session_for_correlation_key("revision-tie") + .unwrap(), + Some(latest.session_id) + ); +} + #[test] fn runtime_session_runner_type_is_available_for_runtime_integration() { struct EmptyApi; @@ -1659,6 +1769,59 @@ async fn turn_steer_event_is_sequenced_after_acceptance() { assert!(steer_index > 0); } +#[tokio::test] +async fn concurrent_steer_and_completion_preserve_sequence_and_terminal_order() { + for _ in 0..16 { + let (started_tx, started_rx) = mpsc::sync_channel(1); + let (release_tx, release_rx) = mpsc::sync_channel(1); + let runner = Arc::new(GateRunner { + started: Mutex::new(Some(started_tx)), + release: Mutex::new(release_rx), + }); + let (_temp, _store, router) = setup_with(runner); + let session = router.create_session(None, None).unwrap(); + let invocation = router.prompt(&session.session_id, "initial", None).unwrap(); + let invocation_id = invocation.invocation_id.clone(); + started_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + + let barrier = Arc::new(Barrier::new(3)); + let steer_router = router.clone(); + let steer_session_id = session.session_id.clone(); + let steer_invocation_id = invocation_id.clone(); + let steer_barrier = Arc::clone(&barrier); + let steer_thread = thread::spawn(move || { + steer_barrier.wait(); + steer_router + .steer(&steer_session_id, &steer_invocation_id, "racing steer") + .is_ok() + }); + let finish_barrier = Arc::clone(&barrier); + let finish_thread = thread::spawn(move || { + finish_barrier.wait(); + release_tx.send(()).unwrap(); + }); + barrier.wait(); + + let steer_accepted = steer_thread.join().unwrap(); + finish_thread.join().unwrap(); + let events = collect_events(invocation).await; + assert!(events + .windows(2) + .all(|pair| pair[0].sequence < pair[1].sequence)); + assert_eq!( + events.last().unwrap().event_type, + SessionEventType::TurnEnded + ); + let steer_index = events + .iter() + .position(|event| event.event_type == SessionEventType::TurnSteered); + assert_eq!(steer_accepted, steer_index.is_some()); + if let Some(index) = steer_index { + assert!(index < events.len() - 1); + } + } +} + #[test] fn session_page_types_round_trip_as_json() { let page = sdk::SessionPage {