Route rollout reads through the canonical JSON decoder (#42378)

## Why

Directly deserializing the flattened `RolloutLine` envelope can reject nested
decimal values, preventing affected paginated sessions from resuming.

## What changed

- Add canonical string, byte, and reverse-scanner helpers that decode rollout
  records through `serde_json::Value` before decoding the flattened item.
- Route rollout readers across session discovery, history, migration, search,
  thread storage, and transcript previews through those helpers.
- Remove `Deserialize` from `RolloutLine` so new readers cannot bypass the
  canonical persistence decoder.

## Testing

Add coverage that resumes a paginated rollout after a token-count record with
a decimal rate-limit value and verifies that ordinal sequencing continues.

GitOrigin-RevId: 49abac1e0751c073daa5a93a840d8a483fd2d013
This commit is contained in:
Owen Lin
2026-09-02 22:45:19 +00:00
committed by copyberry
parent e1d0ef995f
commit 69cebb5d15
49 changed files with 184 additions and 114 deletions
@@ -7,7 +7,6 @@ use codex_rollout::ModelContextScan;
use codex_rollout::ModelContextScanProgress;
use codex_rollout::ReverseJsonlScanner;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutLine;
use codex_rollout::ScanOutcome;
use super::LocalThreadStore;
@@ -137,7 +136,7 @@ fn scan_model_context_from_lineage_blocking(
Some(end_byte_offset) => ReverseJsonlScanner::new_at(file, end_byte_offset)?,
None => ReverseJsonlScanner::new(file)?,
};
while let Some(outcome) = scanner.scan_next::<RolloutLine>()? {
while let Some(outcome) = scanner.scan_next_rollout_line()? {
let ScanOutcome::Parsed(line) = outcome else {
continue;
};
@@ -599,8 +599,8 @@ fn rollout_end_byte_offset(path: &Path, end_ordinal_exclusive: u64) -> u64 {
let contents = std::fs::read(path).expect("read rollout");
let mut byte_offset = 0_u64;
for line in contents.split_inclusive(|byte| *byte == b'\n') {
let parsed: RolloutLine =
serde_json::from_slice(line).expect("parse rollout line for byte offset");
let parsed = codex_rollout::parse_rollout_line_bytes(line)
.expect("parse rollout line for byte offset");
if parsed.ordinal == Some(end_ordinal_exclusive) {
return byte_offset;
}
@@ -327,7 +327,7 @@ fn rollout_end_byte_offset(path: &Path, end_ordinal_exclusive: u64) -> u64 {
let end_byte_offset = bytes
.split_inclusive(|byte| *byte == b'\n')
.take_while(|line| {
serde_json::from_slice::<RolloutLine>(line)
codex_rollout::parse_rollout_line_bytes(line)
.expect("parse rollout fixture")
.ordinal
.expect("paginated rollout ordinal")
@@ -39,7 +39,7 @@ pub(super) fn parse_legacy_rollout_value(mut value: Value) -> Result<Option<Roll
normalize_legacy_rate_limit_resets(&mut value);
normalize_legacy_review_entry(&mut value);
normalize_legacy_command_cwd(&mut value)?;
serde_json::from_value(value)
codex_rollout::decode_rollout_line(value)
.map(Some)
.map_err(|error| error.to_string())
}
@@ -23,12 +23,12 @@ fn parses_legacy_numeric_event_payloads_through_value() {
"info": null,
"rate_limits": {
"primary": {
"used_percent": 2,
"used_percent": 0.0,
"window_minutes": 300,
"resets_at": 1_770_414_841,
},
"secondary": {
"used_percent": 12,
"used_percent": 12.5,
"window_minutes": 10_080,
"resets_at": 1_770_698_702,
},
@@ -23,7 +23,6 @@ use std::os::unix::fs::PermissionsExt;
use codex_protocol::ThreadId;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutLine;
use tokio::fs::File;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
@@ -170,7 +169,7 @@ pub(super) async fn rewrite_subagent_history_boundary(
.read_until(b'\n', &mut head_bytes)
.await
.map_err(migration_error)?;
let mut head = serde_json::from_slice::<RolloutLine>(&head_bytes).map_err(migration_error)?;
let mut head = codex_rollout::parse_rollout_line_bytes(&head_bytes).map_err(migration_error)?;
let RolloutItem::SessionMeta(session_meta) = &mut head.item else {
return Err(migration_error(
"staged rollout head is not session metadata",
@@ -259,7 +259,7 @@ fn read_rollout(path: &Path) -> Vec<RolloutLine> {
fs::read_to_string(path)
.expect("read migrated rollout")
.lines()
.map(|line| serde_json::from_str(line).expect("parse migrated rollout"))
.map(|line| codex_rollout::parse_rollout_line(line).expect("parse migrated rollout"))
.collect()
}
@@ -1392,7 +1392,7 @@ fn rollout_end_byte_offset(path: &std::path::Path, end_ordinal_exclusive: u64) -
let end_byte_offset = bytes
.split_inclusive(|byte| *byte == b'\n')
.take_while(|line| {
serde_json::from_slice::<RolloutLine>(line)
codex_rollout::parse_rollout_line_bytes(line)
.expect("parse rollout fixture")
.ordinal
.expect("paginated rollout ordinal")
@@ -504,7 +504,7 @@ async fn paginated_realtime_items_materialize_separately_in_rollout_order() {
let expected_rows = fs::read_to_string(rollout_path.as_path())
.expect("read canonical rollout")
.lines()
.map(|line| serde_json::from_str::<RolloutLine>(line).expect("parse rollout line"))
.map(|line| codex_rollout::parse_rollout_line(line).expect("parse rollout line"))
.filter_map(|line| match line.item {
RolloutItem::RealtimeItem(item) => Some((
item.id,
@@ -2675,7 +2675,7 @@ fn rollout_line_byte_offsets(path: &std::path::Path, ordinal: u64) -> (i64, i64)
let mut start_byte_offset = 0;
for line in bytes.split_inclusive(|byte| *byte == b'\n') {
let end_byte_offset = start_byte_offset + line.len();
if serde_json::from_slice::<RolloutLine>(line)
if codex_rollout::parse_rollout_line_bytes(line)
.ok()
.and_then(|line| line.ordinal)
== Some(ordinal)