diff --git a/codex-rs/core/src/alarms.rs b/codex-rs/core/src/alarms.rs index 72fc902e62..3ebe3834cb 100644 --- a/codex-rs/core/src/alarms.rs +++ b/codex-rs/core/src/alarms.rs @@ -484,15 +484,43 @@ pub(crate) async fn write_alarm_sidecar( tmp_path.display() ) })?; - tokio::fs::rename(&tmp_path, &sidecar_path) - .await - .map_err(|err| { - format!( - "failed to atomically replace alarm sidecar `{}`: {err}", - sidecar_path.display() - ) - })?; - Ok(()) + match tokio::fs::rename(&tmp_path, &sidecar_path).await { + Ok(()) => Ok(()), + Err(initial_error) => { + #[cfg(target_os = "windows")] + { + match tokio::fs::remove_file(&sidecar_path).await { + Ok(()) => { + tokio::fs::rename(&tmp_path, &sidecar_path) + .await + .map_err(|err| { + format!( + "failed to replace alarm sidecar `{}` with `{}`: {err}", + sidecar_path.display(), + tmp_path.display() + ) + })?; + return Ok(()); + } + Err(err) if err.kind() == ErrorKind::NotFound => {} + Err(err) => { + let _ = tokio::fs::remove_file(&tmp_path).await; + return Err(format!( + "failed to remove existing alarm sidecar `{}` before replace: {err}", + sidecar_path.display() + )); + } + } + } + + let _ = tokio::fs::remove_file(&tmp_path).await; + Err(format!( + "failed to atomically replace alarm sidecar `{}` with `{}`: {initial_error}", + sidecar_path.display(), + tmp_path.display() + )) + } + } } fn parse_duration_literal(raw: &str) -> Option { @@ -792,4 +820,48 @@ mod tests { tempdir.path().join("rollout.jsonl.alarms.json") ); } + + #[tokio::test] + async fn alarm_sidecar_overwrites_existing_file() { + let tempdir = TempDir::new().expect("tempdir"); + let rollout_path = tempdir.path().join("rollout.jsonl"); + let original = vec![PersistedAlarm { + alarm: super::ThreadAlarm { + id: "alarm-1".to_string(), + cron_expression: "@after-turn".to_string(), + prompt: "run tests".to_string(), + run_once: false, + delivery: AlarmDelivery::AfterTurn, + created_at: 1, + next_run_at: None, + last_run_at: None, + }, + pending_run: true, + }]; + let replacement = vec![PersistedAlarm { + alarm: super::ThreadAlarm { + id: "alarm-2".to_string(), + cron_expression: "@after-turn".to_string(), + prompt: "run different tests".to_string(), + run_once: true, + delivery: AlarmDelivery::SteerCurrentTurn, + created_at: 2, + next_run_at: None, + last_run_at: Some(3), + }, + pending_run: false, + }]; + + write_alarm_sidecar(&rollout_path, &original) + .await + .expect("write original sidecar"); + write_alarm_sidecar(&rollout_path, &replacement) + .await + .expect("overwrite sidecar"); + + let loaded = load_alarm_sidecar(&rollout_path) + .await + .expect("load overwritten sidecar"); + assert_eq!(loaded, replacement); + } } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index b307b25ae4..79909c7c86 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2018,7 +2018,6 @@ impl Session { let mut guard = network_policy_decider_session.write().await; *guard = Arc::downgrade(&sess); } - sess.restore_alarms_from_sidecar().await; // Dispatch the SessionConfiguredEvent first and then report any errors. // If resuming, include converted initial messages in the payload so UIs can render them immediately. let initial_messages = initial_history.get_event_msgs(); @@ -2047,6 +2046,7 @@ impl Session { for event in events { sess.send_event_raw(event).await; } + sess.restore_alarms_from_sidecar().await; // Start the watcher after SessionConfigured so it cannot emit earlier events. sess.start_skills_watcher_listener(); @@ -2899,6 +2899,9 @@ impl Session { } async fn restore_alarms_from_sidecar(self: &Arc) { + if !self.features.enabled(Feature::AlarmScheduler) { + return; + } let Some(rollout_path) = self.current_rollout_path().await else { return; };