From 2bd8727a0c07655fafeae0e4b6fe4c2b4741b399 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Thu, 13 Aug 2026 15:57:35 +0000 Subject: [PATCH] Preserve floating-point values when decoding rollout lines (#38399) ## Why Serde's generic buffering for flattened and internally tagged fields is incompatible with `serde_json`'s `arbitrary_precision` representation. As a result, persisted rollout items containing nested floating-point values can fail to decode. ## What changed - Decode the rollout envelope at the JSON persistence boundary before deserializing the item payload. - Use the shared decoder for both session resume and thread-history projection. - Cover token-count floats, response-item metadata, field ordering, and ignored metadata in the decoder compatibility test. GitOrigin-RevId: 59c7f6fbcf83162b5ed6980a69381fe45fc20a0a --- codex-rs/rollout/src/lib.rs | 34 +++++++++++++++++++ codex-rs/rollout/src/recorder.rs | 3 +- codex-rs/rollout/src/tests.rs | 24 +++++++++++++ .../local/thread_history_materialization.rs | 3 +- 4 files changed, 60 insertions(+), 4 deletions(-) diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 196e5f42f6..9f38cd569a 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -3,6 +3,8 @@ use std::sync::LazyLock; use codex_protocol::protocol::SessionSource; +use serde::de::Error as _; +use serde_json::Value; pub(crate) mod compression; pub(crate) mod config; @@ -30,6 +32,38 @@ pub use codex_history::RolloutItem; pub use codex_history::RolloutLine; pub(crate) use codex_protocol::protocol; +/// Decodes a persisted rollout record without Serde's flattened-envelope buffering. +/// +/// With `serde_json/arbitrary_precision`, Serde's generic buffer cannot replay +/// floating-point values nested inside flattened or internally tagged fields: +/// https://github.com/serde-rs/json/issues/721 +/// https://github.com/serde-rs/serde/issues/1183 +/// +/// Keep this JSON-specific workaround at the persistence boundary so history +/// remains format-neutral and resume and projection use the same item decoder. +/// Remove it once Serde supports format-specific buffering. +pub fn decode_rollout_line(value: Value) -> serde_json::Result { + let Value::Object(mut fields) = value else { + return serde_json::from_value(value); + }; + let timestamp = fields + .remove("timestamp") + .ok_or_else(|| serde_json::Error::missing_field("timestamp")) + .and_then(serde_json::from_value)?; + let ordinal = fields + .remove("ordinal") + .map(serde_json::from_value::>) + .transpose()? + .flatten(); + let item = serde_json::from_value(Value::Object(fields))?; + + Ok(RolloutLine { + timestamp, + ordinal, + item, + }) +} + pub const SESSIONS_SUBDIR: &str = "sessions"; pub const ARCHIVED_SESSIONS_SUBDIR: &str = "archived_sessions"; pub static INTERACTIVE_SESSION_SOURCES: LazyLock> = LazyLock::new(|| { diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 93a3c294da..b6fba6b760 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -57,7 +57,6 @@ use super::session_index::find_thread_names_by_ids; use crate::InitialHistory; use crate::ResumedHistory; use crate::RolloutItem; -use crate::RolloutLine; use crate::config::RolloutConfigView; use crate::state_db; use crate::state_db::StateDbHandle; @@ -1035,7 +1034,7 @@ impl RolloutRecorder { reject_unknown_thread_history_mode(&value)?; } - let rollout_line = match serde_json::from_value::(value) { + let rollout_line = match crate::decode_rollout_line(value) { Ok(rollout_line) => rollout_line, Err(e) => { trace!("failed to parse rollout line: {e}"); diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index deb3f0e04b..3dab8585fd 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -49,6 +49,30 @@ use codex_protocol::protocol::UserMessageEvent; const NO_SOURCE_FILTER: &[SessionSource] = &[]; const TEST_PROVIDER: &str = "test-provider"; +#[test] +fn rollout_line_decoder_preserves_canonical_json_compatibility() -> Result<()> { + let cases = [ + r#"{"timestamp":"2025-01-03T12:00:00.000Z","ordinal":7,"type":"event_msg","payload":{"type":"token_count","info":null,"rate_limits":{"limit_id":null,"limit_name":null,"primary":{"used_percent":0.0,"window_minutes":60,"resets_at":1800000000},"secondary":{"used_percent":12.5,"window_minutes":10080,"resets_at":1800100000},"credits":null,"individual_limit":null,"spend_control_reached":null,"plan_type":null,"rate_limit_reached_type":null}}}"#, + r#"{"metadata":{"client_authored":true},"payload":{"type":"message","role":"developer","content":[{"type":"input_text","text":"hello"}]},"type":"response_item","ordinal":9,"timestamp":"2025-01-03T12:00:00.000Z"}"#, + r#"{"timestamp":"2025-01-03T12:00:00.000Z","ordinal":10,"type":"event_msg","payload":{"type":"warning","message":"hello"},"metadata":"ignored"}"#, + ]; + + for encoded in cases { + let value = serde_json::from_str::(encoded)?; + let decoded = crate::decode_rollout_line(value.clone())?; + let mut expected = value; + if expected["type"] != "response_item" { + expected + .as_object_mut() + .expect("rollout object") + .remove("metadata"); + } + assert_eq!(serde_json::to_value(decoded)?, expected); + } + + Ok(()) +} + fn provider_vec(providers: &[&str]) -> Vec { providers .iter() 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 2901d4a9b5..e110ecc2f4 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -5,7 +5,6 @@ use chrono::DateTime; use codex_app_server_protocol::ThreadHistoryChangeSet; use codex_app_server_protocol::project_rollout_line; use codex_protocol::ThreadId; -use codex_rollout::RolloutLine; use tokio::io::AsyncReadExt; use tokio::io::AsyncSeekExt; use tracing::warn; @@ -150,7 +149,7 @@ async fn read_projection_steps( } }; let value_ordinal = value.get("ordinal").and_then(serde_json::Value::as_u64); - let line = match serde_json::from_value::(value) { + let line = match codex_rollout::decode_rollout_line(value) { Ok(line) => Some(line), Err(err) => { warn!(