diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index b20cc17989..c89d1b28b2 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/codex-rs/thread-store/src/local/thread_history.rs @@ -21,16 +21,27 @@ pub(super) use turn_lookup::find_visible_turn; /// A valid complete rollout line with its absolute byte span in durable JSONL. /// -/// `start_byte_offset..end_byte_offset` includes the terminating newline. Blank and rejected -/// lines do not produce a value here, but still advance later spans. +/// `start_byte_offset..end_byte_offset` includes the terminating newline. pub(super) struct ProjectedRolloutLine { pub ordinal: u64, pub start_byte_offset: u64, pub end_byte_offset: u64, - pub created_at_ms: i64, + pub fallback_created_at_ms: Option, pub changes: ThreadHistoryChangeSet, } +/// One ordered update to apply while advancing a rollout projection checkpoint. +/// +/// Skipped ordinal ranges keep the byte and ordinal checkpoints describing the same durable +/// prefix even when a complete rollout line cannot be projected. +pub(super) enum RolloutProjectionStep { + Line(ProjectedRolloutLine), + SkippedOrdinalRange { + start_ordinal: u64, + end_ordinal_exclusive: u64, + }, +} + pub(super) struct RolloutProjectionState { pub next_byte_offset: u64, pub next_ordinal: u64, @@ -91,7 +102,7 @@ pub(super) async fn apply_projection( start_offset: u64, next_offset: u64, initial_ordinal: u64, - projections: Vec, + projections: Vec, ) -> ThreadStoreResult<()> { let pool = store.thread_history_db().await?; // Write the projected rows and advance the JSONL offset and ordinal in one transaction. If @@ -123,29 +134,57 @@ WHERE thread_id = ? } for projection in projections { - let ordinal = sqlite_integer(projection.ordinal, "rollout ordinal")?; - if ordinal != next_ordinal { - return Err(ThreadStoreError::Internal { - message: format!( - "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}" - ), - }); + match projection { + RolloutProjectionStep::Line(projection) => { + let ordinal = sqlite_integer(projection.ordinal, "rollout ordinal")?; + if ordinal != next_ordinal { + return Err(ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}" + ), + }); + } + apply_change_set( + &mut transaction, + thread_id.as_str(), + ordinal, + sqlite_integer(projection.start_byte_offset, "rollout byte offset")?, + sqlite_integer(projection.end_byte_offset, "rollout byte offset")?, + projection.fallback_created_at_ms, + projection.changes, + ) + .await?; + next_ordinal = + next_ordinal + .checked_add(1) + .ok_or_else(|| ThreadStoreError::Internal { + message: "rollout ordinal exceeds SQLite integer range".to_string(), + })?; + } + RolloutProjectionStep::SkippedOrdinalRange { + start_ordinal, + end_ordinal_exclusive, + } => { + let start_ordinal = sqlite_integer(start_ordinal, "rollout ordinal")?; + if start_ordinal != next_ordinal { + return Err(ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {start_ordinal}" + ), + }); + } + let end_ordinal_exclusive = + sqlite_integer(end_ordinal_exclusive, "rollout ordinal")?; + if end_ordinal_exclusive <= start_ordinal { + return Err(ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} has an empty skipped ordinal range" + ), + }); + } + next_ordinal = end_ordinal_exclusive; + } } - apply_change_set( - &mut transaction, - thread_id.as_str(), - ordinal, - sqlite_integer(projection.start_byte_offset, "rollout byte offset")?, - sqlite_integer(projection.end_byte_offset, "rollout byte offset")?, - projection.created_at_ms, - projection.changes, - ) - .await?; - next_ordinal = next_ordinal - .checked_add(1) - .ok_or_else(|| ThreadStoreError::Internal { - message: "rollout ordinal exceeds SQLite integer range".to_string(), - })?; } sqlx::query( @@ -214,7 +253,7 @@ async fn apply_change_set( rollout_ordinal: i64, rollout_byte_offset: i64, rollout_end_byte_offset: i64, - created_at_ms: i64, + fallback_created_at_ms: Option, changes: ThreadHistoryChangeSet, ) -> ThreadStoreResult<()> { for turn in changes.changed_turns { @@ -342,6 +381,14 @@ WHERE thread_id = ? } for item in changes.changed_items { + let created_at_ms = + item.started_at_ms + .or(fallback_created_at_ms) + .ok_or_else(|| ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} is missing an item creation timestamp" + ), + })?; let item_id = item.item.id().to_string(); let item_json = serde_json::to_string(&item.item).map_err(thread_history_error)?; // Completed items are immutable: local producers emit ItemCompleted exactly once per diff --git a/codex-rs/thread-store/src/local/thread_history_materialization.rs b/codex-rs/thread-store/src/local/thread_history_materialization.rs index 496f463f7c..9dced72551 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -12,15 +12,10 @@ use tracing::warn; use super::LocalThreadStore; use super::thread_history::ProjectedRolloutLine; +use super::thread_history::RolloutProjectionStep; use crate::ThreadStoreError; use crate::ThreadStoreResult; -struct CompleteRolloutLine { - line: RolloutLine, - start_byte_offset: u64, - end_byte_offset: u64, -} - pub(super) async fn materialize_to_sqlite( store: &LocalThreadStore, thread_id: ThreadId, @@ -29,12 +24,15 @@ pub(super) async fn materialize_to_sqlite( if store.state_db.is_none() { return Ok(()); } - let start_offset = super::thread_history::projection_state(store, thread_id) - .await? + let projection_state = super::thread_history::projection_state(store, thread_id).await?; + let start_offset = projection_state + .as_ref() .map_or(0, |state| state.next_byte_offset); - let (lines, next_offset) = read_complete_rollout_lines(rollout_path, start_offset).await?; - // Empty valid records can still consume bytes through blank or rejected complete lines. - if lines.is_empty() && start_offset == next_offset { + if projection_state.is_none() + && !tokio::fs::try_exists(rollout_path) + .await + .map_err(thread_store_io_error)? + { return Ok(()); } let session_meta = codex_rollout::read_session_meta_line(rollout_path) @@ -45,31 +43,21 @@ pub(super) async fn materialize_to_sqlite( .history_base .map_or(0, |base| base.end_ordinal_exclusive); let subagent_history_start_ordinal = session_meta.subagent_history_start_ordinal; - - let projections = lines - .iter() - .map(|record| { - let line = &record.line; - let ordinal = line.ordinal.ok_or_else(|| ThreadStoreError::Internal { - message: format!("paginated rollout line for {thread_id} is missing an ordinal"), - })?; - let created_at_ms = DateTime::parse_from_rfc3339(line.timestamp.as_str()) - .map(|timestamp| timestamp.timestamp_millis()) - .map_err(thread_history_error)?; - let changes = if subagent_history_start_ordinal.is_some_and(|start| ordinal < start) { - ThreadHistoryChangeSet::default() - } else { - project_rollout_line(line) - }; - Ok(ProjectedRolloutLine { - ordinal, - start_byte_offset: record.start_byte_offset, - end_byte_offset: record.end_byte_offset, - created_at_ms, - changes, - }) - }) - .collect::>>()?; + let expected_ordinal = projection_state + .as_ref() + .map_or(initial_ordinal, |state| state.next_ordinal); + let (projections, next_offset) = read_projection_steps( + rollout_path, + start_offset, + expected_ordinal, + thread_id, + subagent_history_start_ordinal, + ) + .await?; + // Empty valid records can still consume bytes through blank complete lines. + if projections.is_empty() && start_offset == next_offset { + return Ok(()); + } super::thread_history::apply_projection( store, thread_id, @@ -81,11 +69,14 @@ pub(super) async fn materialize_to_sqlite( .await } -async fn read_complete_rollout_lines( +async fn read_projection_steps( rollout_path: &Path, start_offset: u64, -) -> ThreadStoreResult<(Vec, u64)> { - let next_offset = match tokio::fs::metadata(rollout_path).await { + expected_ordinal: u64, + thread_id: ThreadId, + subagent_history_start_ordinal: Option, +) -> ThreadStoreResult<(Vec, u64)> { + let file_end_offset = match tokio::fs::metadata(rollout_path).await { Ok(metadata) => metadata.len(), Err(err) if err.kind() == std::io::ErrorKind::NotFound && start_offset == 0 => { return Ok((Vec::new(), 0)); @@ -93,7 +84,7 @@ async fn read_complete_rollout_lines( Err(err) => return Err(thread_store_io_error(err)), }; let byte_count = - next_offset + file_end_offset .checked_sub(start_offset) .ok_or_else(|| ThreadStoreError::Internal { message: "durable rollout shrank before projection".to_string(), @@ -117,19 +108,13 @@ async fn read_complete_rollout_lines( .iter() .rposition(|byte| *byte == b'\n') .map_or(0, |index| index + 1); - let next_offset = start_offset - .checked_add(u64::try_from(complete_byte_count).map_err(|_| { - ThreadStoreError::Internal { - message: "durable rollout append exceeds addressable memory".to_string(), - } - })?) - .ok_or_else(|| ThreadStoreError::Internal { - message: "durable rollout byte offset overflow".to_string(), - })?; - let mut lines = Vec::new(); + let mut projections = Vec::new(); + let mut next_ordinal = expected_ordinal; + let mut next_offset = start_offset; + let mut pending_rejected_line_count = 0; let mut line_start_offset = start_offset; - // Preserve each complete physical line's trailing newline so byte offsets advance through - // every durable byte, including blank or rejected lines that do not project a row. + // Keep rejected lines pending until a later valid ordinal proves whether they consumed history. + // This lets a same-ordinal retry replace a failed write without advancing only one checkpoint. for line_bytes in bytes[..complete_byte_count].split_inclusive(|byte| *byte == b'\n') { let line_end_offset = line_start_offset .checked_add(u64::try_from(line_bytes.len()).map_err(|_| { @@ -140,35 +125,153 @@ async fn read_complete_rollout_lines( .ok_or_else(|| ThreadStoreError::Internal { message: "durable rollout byte offset overflow".to_string(), })?; - // Blank physical lines consume bytes but are not rollout records. - if !line_bytes.iter().all(u8::is_ascii_whitespace) { - match serde_json::from_slice::(line_bytes) - .and_then(serde_json::from_value::) - { - Ok(line) => lines.push(CompleteRolloutLine { - line, - start_byte_offset: line_start_offset, - end_byte_offset: line_end_offset, - }), + if line_bytes.iter().all(u8::is_ascii_whitespace) { + if pending_rejected_line_count == 0 { + next_offset = line_end_offset; + } + line_start_offset = line_end_offset; + continue; + } + let value = match serde_json::from_slice::(line_bytes) { + Ok(value) => value, + Err(err) => { + warn!( + thread_id = %thread_id, + rollout_path = %rollout_path.display(), + line_start_byte_offset = line_start_offset, + line_end_byte_offset = line_end_offset, + expected_ordinal = next_ordinal, + error = %err, + "deferring rejected rollout line until a later ordinal resolves it" + ); + pending_rejected_line_count += 1; + line_start_offset = line_end_offset; + continue; + } + }; + let value_ordinal = value.get("ordinal").and_then(serde_json::Value::as_u64); + let line = match serde_json::from_value::(value) { + Ok(line) => Some(line), + Err(err) => { + warn!( + thread_id = %thread_id, + rollout_path = %rollout_path.display(), + line_start_byte_offset = line_start_offset, + line_end_byte_offset = line_end_offset, + expected_ordinal = next_ordinal, + line_ordinal = ?value_ordinal, + error = %err, + "deferring unknown rollout line until a later ordinal resolves it" + ); + None + } + }; + let ordinal = match line + .as_ref() + .and_then(|line| line.ordinal) + .or(value_ordinal) + { + Some(ordinal) => ordinal, + None if line.is_none() => { + pending_rejected_line_count += 1; + line_start_offset = line_end_offset; + continue; + } + None => { + return Err(ThreadStoreError::Internal { + message: format!( + "paginated rollout line for {thread_id} is missing an ordinal" + ), + }); + } + }; + if ordinal < next_ordinal { + return Err(ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}" + ), + }); + } + let Some(line) = line else { + pending_rejected_line_count += 1; + line_start_offset = line_end_offset; + continue; + }; + let skipped_ordinal_count = ordinal - next_ordinal; + if skipped_ordinal_count > pending_rejected_line_count { + return Err(ThreadStoreError::Internal { + message: format!( + "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}; {pending_rejected_line_count} rejected rollout lines cannot cover that gap" + ), + }); + } + let changes = if subagent_history_start_ordinal.is_some_and(|start| ordinal < start) { + ThreadHistoryChangeSet::default() + } else { + project_rollout_line(&line) + }; + let fallback_created_at_ms = if changes + .changed_items + .iter() + .any(|item| item.started_at_ms.is_none()) + { + match DateTime::parse_from_rfc3339(line.timestamp.as_str()) { + Ok(timestamp) => Some(timestamp.timestamp_millis()), Err(err) => { - // A failed append can leave a partial record behind. The rollout writer - // repairs its newline before retrying, so skip rejected lines just like the - // canonical rollout loader and keep projecting the valid retry that follows. warn!( - "skipping rejected rollout line while projecting {rollout_path:?}: {err}" + thread_id = %thread_id, + rollout_path = %rollout_path.display(), + line_start_byte_offset = line_start_offset, + line_end_byte_offset = line_end_offset, + expected_ordinal = next_ordinal, + line_ordinal = ordinal, + error = %err, + "deferring rollout line with invalid timestamp until a later ordinal resolves it" ); + pending_rejected_line_count += 1; + line_start_offset = line_end_offset; + continue; } } + } else { + None + }; + if skipped_ordinal_count > 0 { + warn!( + thread_id = %thread_id, + rollout_path = %rollout_path.display(), + line_start_byte_offset = line_start_offset, + line_end_byte_offset = line_end_offset, + expected_ordinal = next_ordinal, + line_ordinal = ordinal, + skipped_ordinal_start = next_ordinal, + skipped_ordinal_end_exclusive = ordinal, + "skipping rollout ordinal range after rejected lines" + ); + projections.push(RolloutProjectionStep::SkippedOrdinalRange { + start_ordinal: next_ordinal, + end_ordinal_exclusive: ordinal, + }); } + pending_rejected_line_count = 0; + let next_line_ordinal = + ordinal + .checked_add(1) + .ok_or_else(|| ThreadStoreError::Internal { + message: "rollout ordinal exceeds SQLite integer range".to_string(), + })?; + projections.push(RolloutProjectionStep::Line(ProjectedRolloutLine { + ordinal, + start_byte_offset: line_start_offset, + end_byte_offset: line_end_offset, + fallback_created_at_ms, + changes, + })); + next_ordinal = next_line_ordinal; + next_offset = line_end_offset; line_start_offset = line_end_offset; } - Ok((lines, next_offset)) -} - -fn thread_history_error(err: impl std::fmt::Display) -> ThreadStoreError { - ThreadStoreError::Internal { - message: format!("failed to project thread history: {err}"), - } + Ok((projections, next_offset)) } fn thread_store_io_error(err: std::io::Error) -> ThreadStoreError { diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index bd58bf0f27..465d401a74 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -1440,7 +1440,6 @@ async fn paginated_projection_accepts_float_rate_limits_and_later_final_answers( .expect("queue unprojected history"); recorder.flush().await.expect("flush unprojected history"); assert_eq!(projection_state(&pool, thread_id).await, checkpoint); - super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) .await .expect("project floating-point rate limits"); @@ -1712,6 +1711,13 @@ async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() { "out of order ordinal", format!("{}\n", rollout_line(Some(2), turn_started("turn-1"))), ), + ( + "gap larger than rejected prefix", + format!( + "{{not json}}\n{}\n", + rollout_line(Some(3), turn_started("turn-1")), + ), + ), ]; for (name, suffix) in cases { let home = TempDir::new().expect("temp dir"); @@ -1852,6 +1858,12 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { .await .expect("persist session metadata"); + let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( + home.path().abs(), + )) + .await + .expect("open thread history db"); + let before = projection_state(&pool, thread_id).await; let rollout_path = store .live_rollout_path(thread_id) .await @@ -1863,6 +1875,14 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { file.write_all(b"\n \t\r\n{not json}\n\xff\n") .expect("append blank and rejected lines"); file.flush().expect("flush rejected line"); + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect("leave rejected tail pending"); + assert_eq!( + projection_state(&pool, thread_id).await, + (before.0 + 5, before.1) + ); + let recorder = store .live_recorders .lock() @@ -1881,11 +1901,6 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { .await .expect("project valid retry after rejected line"); - let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( - home.path().abs(), - )) - .await - .expect("open thread history db"); let (expected_start_byte_offset, _) = rollout_line_byte_offsets(rollout_path.as_path(), /*ordinal*/ 1); let start_byte_offset = sqlx::query_scalar::<_, Option>( @@ -1897,6 +1912,198 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { .await .expect("read projected turn byte offset"); assert_eq!(start_byte_offset, Some(expected_start_byte_offset)); + let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len()) + .expect("rollout length"); + assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 2)); +} + +#[tokio::test] +async fn unprojectable_rollout_lines_wait_for_later_ordinals() { + let unknown_line = |ordinal| { + format!( + concat!( + "{{\"timestamp\":\"2025-01-01T00:00:00.000Z\",\"ordinal\":{ordinal},", + "\"type\":\"future_item\",\"payload\":{{}}}}" + ), + ordinal = ordinal + ) + }; + let cases = [ + ("unknown payload", unknown_line(1), unknown_line(2)), + ( + "structurally invalid JSON", + "{}".to_string(), + "null".to_string(), + ), + ]; + for (name, pending_line, skipped_line) in cases { + let home = TempDir::new().expect("temp dir"); + let store = projection_store(home.path()).await; + let thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + store + .persist_thread(thread_id) + .await + .expect("persist session metadata"); + + let pool = codex_state::open_thread_history_db( + &codex_state::SqliteConfig::new_for_testing(home.path().abs()), + ) + .await + .expect("open thread history db"); + let before = projection_state(&pool, thread_id).await; + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("rollout path"); + append_suffix(rollout_path.as_path(), format!("{pending_line}\n").as_str()); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect("leave unprojectable tail pending"); + assert_eq!(projection_state(&pool, thread_id).await, before, "{name}"); + + append_suffix( + rollout_path.as_path(), + format!( + "{}\n{skipped_line}\n{}\n", + rollout_line(Some(1), turn_started("retry-turn")), + rollout_line(Some(3), turn_started("turn-1")), + ) + .as_str(), + ); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect(name); + + let rollout_len = + i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len()) + .expect("rollout length"); + assert_eq!( + projection_state(&pool, thread_id).await, + (rollout_len, 4), + "{name}" + ); + let turn_ordinals = sqlx::query_as::<_, (String, i64)>( + "SELECT turn_id, rollout_ordinal FROM thread_turns WHERE thread_id = ? ORDER BY rollout_ordinal", + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read projected turns"); + assert_eq!( + turn_ordinals, + vec![("retry-turn".to_string(), 1), ("turn-1".to_string(), 3)], + "{name}" + ); + } +} + +#[tokio::test] +async fn event_timestamps_allow_invalid_rollout_timestamps() { + let home = TempDir::new().expect("temp dir"); + let store = projection_store(home.path()).await; + let thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + store + .persist_thread(thread_id) + .await + .expect("persist session metadata"); + + let invalid_timestamp_line = |ordinal, item| { + rollout_line(Some(ordinal), item).replace("2025-01-01T00:00:00.000Z", "not-a-timestamp") + }; + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("rollout path"); + append_suffix( + rollout_path.as_path(), + format!( + "{}\n{}\n", + invalid_timestamp_line(1, turn_started("turn-1")), + invalid_timestamp_line( + 2, + completed_item( + thread_id, + "turn-1", + agent_message("agent-1", MessagePhase::FinalAnswer), + ), + ), + ) + .as_str(), + ); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect("project records with event timestamps"); + + let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( + home.path().abs(), + )) + .await + .expect("open thread history db"); + let created_at_ms = sqlx::query_scalar::<_, i64>( + "SELECT created_at_ms FROM thread_items WHERE thread_id = ? AND turn_id = ? AND item_id = ?", + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .bind("agent-1") + .fetch_one(&pool) + .await + .expect("read projected item timestamp"); + assert_eq!(created_at_ms, 0); + let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len()) + .expect("rollout length"); + assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 3)); +} + +#[tokio::test] +async fn malformed_rollout_lines_skip_inferred_ordinal_gaps() { + let home = TempDir::new().expect("temp dir"); + let store = projection_store(home.path()).await; + let thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + store + .persist_thread(thread_id) + .await + .expect("persist session metadata"); + + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("rollout path"); + append_suffix( + rollout_path.as_path(), + format!( + "{{not json}}\n{}\n", + rollout_line(Some(2), turn_started("turn-1")) + ) + .as_str(), + ); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect("skip malformed gap and project later history"); + + let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( + home.path().abs(), + )) + .await + .expect("open thread history db"); + let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len()) + .expect("rollout length"); + assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 3)); + let turn_ordinal = sqlx::query_scalar::<_, i64>( + "SELECT rollout_ordinal FROM thread_turns WHERE thread_id = ? AND turn_id = ?", + ) + .bind(thread_id.to_string()) + .bind("turn-1") + .fetch_one(&pool) + .await + .expect("read projected turn"); + assert_eq!(turn_ordinal, 2); } #[tokio::test]