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 989715493b..496f463f7c 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -142,7 +142,9 @@ async fn read_complete_rollout_lines( })?; // 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) { + 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, 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 ae4f4c46f8..bd58bf0f27 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 @@ -17,11 +17,14 @@ use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::ItemCompletedEvent; +use codex_protocol::protocol::RateLimitSnapshot; +use codex_protocol::protocol::RateLimitWindow; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadMemoryMode; +use codex_protocol::protocol::TokenCountEvent; use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; @@ -1361,6 +1364,148 @@ async fn summary_items_use_final_answers_and_ignore_commentary() { ); } +#[tokio::test] +async fn paginated_projection_accepts_float_rate_limits_and_later_final_answers() { + 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"); + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![turn_started("turn-1")], + }) + .await + .expect("append projected turn start"); + + let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( + home.path().abs(), + )) + .await + .expect("open thread history db"); + let checkpoint = projection_state(&pool, thread_id).await; + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("rollout path"); + let token_count = |primary: Option, secondary: Option| { + RolloutItem::EventMsg(EventMsg::TokenCount(TokenCountEvent { + info: None, + rate_limits: Some(RateLimitSnapshot { + limit_id: None, + limit_name: None, + primary: primary.map(|used_percent| RateLimitWindow { + used_percent, + window_minutes: Some(60), + resets_at: Some(1_800_000_000), + }), + secondary: secondary.map(|used_percent| RateLimitWindow { + used_percent, + window_minutes: Some(10_080), + resets_at: Some(1_800_100_000), + }), + credits: None, + individual_limit: None, + spend_control_reached: None, + plan_type: None, + rate_limit_reached_type: None, + }), + })) + }; + + let recorder = store + .live_recorders + .lock() + .await + .get(&thread_id) + .expect("live recorder") + .recorder + .clone(); + recorder + .record_canonical_items(&[ + token_count(Some(0.0), Some(1.0)), + completed_item( + thread_id, + "turn-1", + agent_message("final-1", MessagePhase::FinalAnswer), + ), + token_count(Some(12.5), None), + turn_completed("turn-1"), + ]) + .await + .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"); + + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![ + turn_started("turn-2"), + token_count(None, Some(87.25)), + completed_item( + thread_id, + "turn-2", + agent_message("final-2", MessagePhase::FinalAnswer), + ), + token_count(Some(1e-12), Some(1_000_000.0)), + turn_completed("turn-2"), + ], + }) + .await + .expect("append history after unprojected floating-point rate limits"); + + let final_answers = sqlx::query_as::<_, (String, String, String, i64)>( + r#" +SELECT turns.turn_id, items.item_id, turns.status, items.rollout_ordinal +FROM thread_turns AS turns +JOIN thread_items AS items + ON items.thread_id = turns.thread_id + AND items.turn_id = turns.turn_id + AND items.item_id = turns.final_agent_item_id +WHERE turns.thread_id = ? +ORDER BY turns.rollout_ordinal + "#, + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read projected final answers"); + assert_eq!( + final_answers, + vec![ + ( + "turn-1".to_string(), + "final-1".to_string(), + "completed".to_string(), + 3, + ), + ( + "turn-2".to_string(), + "final-2".to_string(), + "completed".to_string(), + 8, + ), + ] + ); + + let rollout_len = i64::try_from( + fs::metadata(rollout_path.as_path()) + .expect("rollout metadata") + .len(), + ) + .expect("rollout length"); + assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 11)); +} + #[tokio::test] async fn next_write_catches_up_unprojected_durable_suffix() { let home = TempDir::new().expect("temp dir");