Skip to content

Commit 55e916e

Browse files
committed
fix(rollout): decode resume tail through canonical path
1 parent eb10d91 commit 55e916e

2 files changed

Lines changed: 39 additions & 2 deletions

File tree

‎codex-rs/rollout/src/ordinal.rs‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,11 @@ use std::path::Path;
88

99
use codex_protocol::protocol::HistoryPosition;
1010
use codex_protocol::protocol::ThreadHistoryMode;
11+
use serde_json::Value;
1112

1213
use crate::RolloutItem;
1314
use crate::RolloutLine;
15+
use crate::decode_rollout_line;
1416
use crate::reverse_jsonl_scanner::ReverseJsonlScanner;
1517
use crate::reverse_jsonl_scanner::ScanOutcome;
1618

@@ -67,8 +69,11 @@ pub(crate) fn ordinal_state_for_rollout(
6769

6870
let mut scanner = ReverseJsonlScanner::new(file)?;
6971
let record = loop {
70-
match scanner.scan_next::<RolloutLine>()? {
71-
Some(ScanOutcome::Parsed(record)) => break record,
72+
match scanner.scan_next::<Value>()? {
73+
Some(ScanOutcome::Parsed(value)) => match decode_rollout_line(value) {
74+
Ok(record) => break record,
75+
Err(_) => continue,
76+
},
7277
Some(ScanOutcome::Rejected(_)) => continue,
7378
None => {
7479
return Err(io::Error::other(format!(

‎codex-rs/rollout/src/recorder_tests.rs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -995,6 +995,38 @@ async fn resumed_paginated_rollout_continues_after_ordinal_gap() -> std::io::Res
995995
recorder.shutdown().await
996996
}
997997

998+
#[tokio::test]
999+
async fn resumed_paginated_rollout_decodes_float_rate_limit_tail_before_assigning_ordinal()
1000+
-> std::io::Result<()> {
1001+
let home = TempDir::new().expect("temp dir");
1002+
let config = test_config(home.path());
1003+
let rollout_path = home.path().join("rollout.jsonl");
1004+
write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?;
1005+
1006+
let token_count = r#"{"timestamp":"2026-07-09T00:00:05Z","ordinal":5,"type":"event_msg","payload":{"type":"token_count","info":null,"rate_limits":{"limit_id":null,"limit_name":null,"primary":{"used_percent":49.0,"window_minutes":60,"resets_at":1800000000},"secondary":null,"credits":null,"individual_limit":null,"spend_control_reached":null,"plan_type":null,"rate_limit_reached_type":null}}}"#;
1007+
assert!(serde_json::from_str::<RolloutLine>(token_count).is_err());
1008+
let mut file = fs::OpenOptions::new().append(true).open(&rollout_path)?;
1009+
writeln!(file, "{token_count}")?;
1010+
drop(file);
1011+
1012+
let recorder =
1013+
RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?;
1014+
recorder
1015+
.record_canonical_items(&[agent_message_item("after-resume")])
1016+
.await?;
1017+
recorder.flush().await?;
1018+
1019+
let ordinals = fs::read_to_string(&rollout_path)?
1020+
.lines()
1021+
.map(serde_json::from_str::<serde_json::Value>)
1022+
.collect::<Result<Vec<_>, _>>()?
1023+
.into_iter()
1024+
.map(|value| value.get("ordinal").and_then(serde_json::Value::as_u64))
1025+
.collect::<Vec<_>>();
1026+
assert_eq!(ordinals, vec![Some(0), Some(4), Some(5), Some(6)]);
1027+
recorder.shutdown().await
1028+
}
1029+
9981030
#[tokio::test]
9991031
async fn resumed_paginated_rollout_repairs_unsafe_tail() -> std::io::Result<()> {
10001032
let valid_unterminated = serde_json::to_string(&RolloutLine {

0 commit comments

Comments
 (0)