codex/codex-rs/thread-store/src/local/rollout_migration_tests.rs

use std::fs;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;

#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;

use codex_app_server_protocol::build_turns_from_rollout_items;
use codex_extension_items::ExtensionItem;
use codex_extension_items::image_generation::GeneratedImageRef;
use codex_extension_items::image_generation::ImageGenerationFailure;
use codex_extension_items::image_generation::ImageGenerationInputImage;
use codex_extension_items::image_generation::ImageGenerationItem;
use codex_protocol::AgentPath;
use codex_protocol::ResponseItemId;
use codex_protocol::ThreadId;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::items::ReasoningItem;
use codex_protocol::items::TurnItem;
use codex_protocol::mcp::McpResourceOrigin;
use codex_protocol::mcp::McpResourceOriginCheckpoint;
use codex_protocol::models::AudioReference;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ContentItemKind;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::ImageReference;
use codex_protocol::models::InternalChatMessageMetadataPassthrough;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ImageGenerationEndEvent;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UserMessageAudioKind;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::protocol::UserMessageImageKind;
use codex_protocol::user_input::UserInput;
use codex_protocol_test_support::uuid_v7_item_id_generator_for_test;
use codex_rollout::CompactedItem;
use codex_rollout::RetainedContextEvent;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutLine;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;

use super::LocalThreadStore;
use super::MigrationLocks;
use super::RolloutMigrationFailureReason;
use super::RolloutMigrationMode;
use super::RolloutMigrationOptions;
use super::RolloutMigrationPaths;
use super::RolloutMigrationProgress;
use super::RolloutMigrationStatus;
use super::compress_rollout_to_path;
use super::decompress_rollout_to_path;
use super::migration_journal_path;
use super::telemetry::RolloutMigrationTrigger;
use super::thread_history;
use super::write_migration_journal;
use crate::ItemSortKey;
use crate::ListItemsParams;
use crate::ListThreadsParams;
use crate::ListTurnsParams;
use crate::LoadThreadHistoryParams;
use crate::SortDirection;
use crate::StoredTurnItemsView;
use crate::ThreadMetadataPatch;
use crate::ThreadSortKey;
use crate::ThreadStore;
use crate::ThreadStoreError;
use crate::TurnPage;
use crate::UpdateThreadMetadataParams;
use crate::local::test_support::test_config;

const TIMESTAMP: &str = "2025-01-03T12:00:00Z";

fn write_rollout(
    home: &Path,
    thread_id: ThreadId,
    source: SessionSource,
    items: Vec<RolloutItem>,
) -> PathBuf {
    write_rollout_with_fork(home, thread_id, source, /*forked_from_id*/ None, items)
}

fn write_rollout_with_fork(
    home: &Path,
    thread_id: ThreadId,
    source: SessionSource,
    forked_from_id: Option<ThreadId>,
    items: Vec<RolloutItem>,
) -> PathBuf {
    let directory = home.join("sessions/2025/01/03");
    fs::create_dir_all(&directory).expect("create rollout directory");
    let path = directory.join(format!("rollout-2025-01-03T12-00-00-{thread_id}.jsonl"));
    let mut file = fs::File::create(&path).expect("create legacy rollout");
    let metadata = SessionMeta {
        session_id: thread_id.into(),
        id: thread_id,
        forked_from_id,
        timestamp: TIMESTAMP.to_string(),
        cwd: home.to_path_buf(),
        originator: "test-originator".to_string(),
        cli_version: "0.0.0".to_string(),
        source,
        model_provider: Some("test-provider".to_string()),
        ..SessionMeta::default()
    };
    let items = std::iter::once(RolloutItem::SessionMeta(Box::new(SessionMetaLine {
        meta: metadata,
        git: None,
    })))
    .chain(items);
    for item in items {
        let line = RolloutLine {
            timestamp: TIMESTAMP.to_string(),
            ordinal: None,
            item,
        };
        writeln!(
            file,
            "{}",
            serde_json::to_string(&line).expect("serialize legacy record")
        )
        .expect("write legacy record");
    }
    path
}

fn move_to_archived(home: &Path, path: PathBuf) -> PathBuf {
    let directory = home.join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR);
    fs::create_dir_all(&directory).expect("create archived rollout directory");
    let archived_path = directory.join(path.file_name().expect("rollout filename"));
    fs::rename(path, &archived_path).expect("archive rollout");
    archived_path
}

fn compress_rollout(path: &Path) -> PathBuf {
    let compressed_path = path.with_extension("jsonl.zst");
    let compressed = zstd::stream::encode_all(
        fs::File::open(path).expect("open rollout"),
        /*level*/ 3,
    )
    .expect("compress rollout");
    fs::write(&compressed_path, compressed).expect("write compressed rollout");
    fs::remove_file(path).expect("remove plain rollout");
    compressed_path
}

fn user_message(text: &str) -> RolloutItem {
    RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
        message: text.to_string(),
        ..UserMessageEvent::default()
    }))
}

fn agent_message(text: &str) -> RolloutItem {
    RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
        message: text.to_string(),
        phase: None,
        memory_citation: None,
        delivery: None,
        questions: None,
    }))
}

fn input_response_message(role: &str, text: &str) -> ResponseItem {
    ResponseItem::Message {
        status: None,
        encrypted_content: None,
        id: None,
        role: role.to_string(),
        content: vec![ContentItem::InputText {
            text: text.to_string(),
        }],
        phase: None,
        internal_chat_message_metadata_passthrough: None,
    }
}

fn rollout_response_item(item: ResponseItem) -> RolloutItem {
    RolloutItem::ResponseItem(item.into())
}

fn exec_completion(turn_id: &str, call_id: &str, output: &str) -> RolloutItem {
    serde_json::from_value(json!({
        "type": "event_msg",
        "payload": {
            "type": "exec_command_end",
            "call_id": call_id,
            "turn_id": turn_id,
            "command": ["echo", "ok"],
            "cwd": "file:///tmp",
            "parsed_cmd": [],
            "source": "agent",
            "stdout": output,
            "stderr": "",
            "aggregated_output": output,
            "exit_code": 0,
            "duration": {"secs": 0, "nanos": 0},
            "formatted_output": output,
            "status": "completed"
        }
    }))
    .expect("build legacy exec completion")
}

fn item_completed(turn_id: &str, item_id: &str) -> RolloutItem {
    RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
        content_token_count: None,
        thread_id: ThreadId::new(),
        turn_id: turn_id.to_string(),
        item: TurnItem::Reasoning(ReasoningItem {
            id: item_id.to_string(),
            summary_text: vec!["summary".to_string()],
            raw_content: Vec::new(),
        }),
        started_at_ms: None,
        completed_at_ms: 1_735_905_601_000,
    }))
}

fn started(turn_id: &str) -> RolloutItem {
    RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
        turn_attribution: None,
        turn_id: turn_id.to_string(),
        root_turn_id: None,
        trace_id: None,
        started_at: Some(1_735_905_600),
        model_context_window: None,
        collaboration_mode_kind: Default::default(),
    }))
}

fn compacted(replacement_history: Vec<ResponseItem>) -> RolloutItem {
    RolloutItem::Compacted(Box::new(CompactedItem {
        message: "checkpoint".to_string(),
        replacement_history: Some(replacement_history.into_iter().map(Into::into).collect()),
        retained_context: None,
        guardian_history: None,
        mcp_resource_origins: None,
        window_number: Some(1),
        first_window_id: None,
        previous_window_id: None,
        window_id: None,
        compaction_response_id: None,
        latest_token_usage_record: None,
        resume_metadata: None,
    }))
}

fn completed(turn_id: &str) -> RolloutItem {
    RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
        root_turn_id: None,
        turn_id: turn_id.to_string(),
        last_agent_message: None,
        error: None,
        started_at: Some(1_735_905_600),
        completed_at: Some(1_735_905_601),
        duration_ms: Some(1_000),
        time_to_first_token_ms: None,
    }))
}

fn read_rollout(path: &Path) -> Vec<RolloutLine> {
    fs::read_to_string(path)
        .expect("read migrated rollout")
        .lines()
        .map(|line| codex_rollout::parse_rollout_line(line).expect("parse migrated rollout"))
        .collect()
}

fn apply_options() -> RolloutMigrationOptions {
    RolloutMigrationOptions {
        mode: RolloutMigrationMode::Apply,
        max_mib_per_second: Some(1024),
        ..RolloutMigrationOptions::default()
    }
}

fn assert_failed_with_reason(
    outcome: &super::RolloutMigrationOutcome,
    failure_reason: RolloutMigrationFailureReason,
) {
    assert_eq!(
        (outcome.status, outcome.failure_reason),
        (RolloutMigrationStatus::Failed, Some(failure_reason))
    );
}

async fn indexed_store(home: &Path) -> LocalThreadStore {
    let config = test_config(home);
    let rollout_config = RolloutConfig {
        codex_home: config.codex_home.clone(),
        sqlite: config.sqlite.clone(),
        cwd: home.to_path_buf(),
        model_provider_id: config.default_model_provider_id.clone(),
        generate_memories: false,
    };
    let state_db = codex_rollout::state_db::try_init(&rollout_config)
        .await
        .expect("backfill legacy thread metadata");
    LocalThreadStore::new(config, Some(state_db))
}

async fn list_active_summary_turns(store: &LocalThreadStore, thread_id: ThreadId) -> TurnPage {
    store
        .list_turns(ListTurnsParams {
            thread_id,
            include_archived: false,
            cursor: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            items_view: StoredTurnItemsView::Summary,
        })
        .await
        .expect("read projected turns")
}

#[tokio::test]
async fn migration_publishes_canonical_projected_history_and_is_idempotent() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let user_event = UserMessageEvent {
        message: "first question".to_string(),
        images: Some(vec!["https://example.com/image.png".to_string()]),
        image_details: vec![Some(ImageDetail::Original)],
        file_ids: Some(vec!["file_123".to_string()]),
        file_id_details: vec![Some(ImageDetail::High)],
        image_order: vec![UserMessageImageKind::File, UserMessageImageKind::Inline],
        audio: Some(vec!["data:audio/wav;base64,YXVkaW8=".into()]),
        audio_references: Some(vec![AudioReference::File {
            file_id: "file_audio".into(),
        }]),
        audio_order: vec![UserMessageAudioKind::File, UserMessageAudioKind::Inline],
        ..Default::default()
    };
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            RolloutItem::EventMsg(EventMsg::UserMessage(user_event)),
            agent_message("first answer"),
        ],
    );
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy rollout");
    assert_eq!(report.outcomes.len(), 1);
    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);

    let lines = read_rollout(&path);
    assert_eq!(
        lines.iter().map(|line| line.ordinal).collect::<Vec<_>>(),
        (0..lines.len() as u64).map(Some).collect::<Vec<_>>()
    );
    assert!(matches!(
        &lines[0].item,
        RolloutItem::SessionMeta(metadata)
            if metadata.meta.history_mode == ThreadHistoryMode::Paginated
                && metadata.meta.id == thread_id
                && metadata.meta.history_base.is_none()
    ));
    assert_eq!(
        lines
            .iter()
            .filter(|line| matches!(line.item, RolloutItem::EventMsg(EventMsg::ItemCompleted(_))))
            .count(),
        2
    );
    let user_item = lines.iter().find_map(|line| match &line.item {
        RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
            item: TurnItem::UserMessage(item),
            ..
        })) => Some(item),
        _ => None,
    });
    let expected_content = vec![
        UserInput::Text {
            text: "first question".to_string(),
            text_elements: Vec::new(),
        },
        UserInput::Image {
            image: ImageReference::File {
                file_id: "file_123".to_string(),
            },
            detail: Some(ImageDetail::High),
        },
        UserInput::Image {
            image: ImageReference::Inline {
                image_url: "https://example.com/image.png".to_string(),
            },
            detail: Some(ImageDetail::Original),
        },
        UserInput::Audio {
            audio: AudioReference::File {
                file_id: "file_audio".into(),
            },
        },
        UserInput::Audio {
            audio: AudioReference::Inline {
                audio_url: "data:audio/wav;base64,YXVkaW8=".into(),
            },
        },
    ];
    assert_eq!(user_item.map(|item| &item.content), Some(&expected_content));

    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(turns.turns.len(), 1);
    assert_eq!(turns.turns[0].items.len(), 2);

    let bytes = fs::read(&path).expect("read first migration");
    let second = store
        .migrate_rollouts(apply_options())
        .await
        .expect("rerun migration");
    assert_eq!(
        second.outcomes[0].status,
        RolloutMigrationStatus::AlreadyPaginated
    );
    assert_eq!(fs::read(&path).expect("read idempotent rollout"), bytes);
}

#[tokio::test]
async fn migration_projects_explicit_and_implicit_legacy_completed_items() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let output = format!("head\n{}\ntail", "x".repeat(128 * 1024));
    let exec = exec_completion("explicit", "call-1", &output);
    let reasoning = serde_json::from_value(json!({
        "type": "event_msg",
        "payload": {"type": "agent_reasoning", "text": "summary"}
    }))
    .expect("build legacy reasoning summary");
    let raw_reasoning = serde_json::from_value(json!({
        "type": "event_msg",
        "payload": {"type": "agent_reasoning_raw_content", "text": "raw"}
    }))
    .expect("build legacy reasoning content");
    let child_id = ThreadId::new();
    let [old_activity, activity] = [None, Some("gpt-5")].map(|model| {
        let mut payload = json!({
            "type": "sub_agent_activity",
            "event_id": if model.is_some() { "spawn" } else { "old-spawn" },
            "kind": "started",
            "agent_thread_id": child_id,
            "agent_path": "/root/worker",
        });
        if let Some(model) = model {
            payload["model"] = json!(model);
            payload["reasoning_effort"] = json!("high");
        }
        serde_json::from_value(json!({"type": "event_msg", "payload": payload}))
            .expect("build legacy activity")
    });
    write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("explicit"),
            exec,
            old_activity,
            activity,
            completed("explicit"),
            reasoning,
            raw_reasoning,
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy completion events");

    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(turns.turns.len(), 2);
    assert_eq!(turns.turns[0].turn_id, "explicit");
    let items = store
        .list_items(ListItemsParams {
            item_ids: None,
            thread_id,
            turn_id: None,
            include_archived: false,
            position: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            sort_key: ItemSortKey::CreatedAtOrdinal,
            after_updated_at_ordinal: None,
            before_updated_at_ordinal: None,
        })
        .await
        .expect("read projected items");
    assert_eq!(items.items.len(), 4);
    assert_eq!(items.items[0].turn_id, "explicit");
    assert_eq!(items.items[3].turn_id, turns.turns[1].turn_id);
    let command: serde_json::Value =
        serde_json::from_slice(&items.items[0].item_json).expect("parse projected command");
    let reasoning: serde_json::Value =
        serde_json::from_slice(&items.items[3].item_json).expect("parse projected reasoning");
    assert_eq!(command["type"], "commandExecution");
    let aggregated_output = command["aggregatedOutput"]
        .as_str()
        .expect("migrated command output");
    assert_eq!(aggregated_output.len(), 64 * 1024);
    assert!(aggregated_output.starts_with("head\n"));
    assert!(aggregated_output.ends_with("\ntail"));
    assert!(aggregated_output.contains("command output truncated for persistence"));
    assert_eq!(reasoning["type"], "reasoning");
    assert_eq!(reasoning["summary"], json!(["summary"]));
    assert_eq!(reasoning["content"], json!(["raw"]));
    for (item, model) in items.items[1..3].iter().zip([None, Some("gpt-5")]) {
        assert_eq!(item.turn_id, "explicit");
        assert_eq!(
            serde_json::from_slice::<serde_json::Value>(&item.item_json).unwrap(),
            json!({
                "type": "subAgentActivity",
                "id": if model.is_some() { "spawn" } else { "old-spawn" },
                "kind": "started",
                "agentThreadId": child_id,
                "agentPath": "/root/worker",
                "model": model,
                "reasoningEffort": model.map(|_| "high"),
            })
        );
    }
}

#[tokio::test]
async fn migration_preserves_image_generation_failure_metadata() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let expected_item = ImageGenerationItem {
        id: "image-call".to_string(),
        status: "failed".to_string(),
        revised_prompt: Some("paint a blue whale".to_string()),
        image: GeneratedImageRef::Inline {
            result: String::new(),
        },
        transparent_background: None,
        failure: Some(ImageGenerationFailure::UsageLimitExceeded {
            limit_id: "image_gen".to_string(),
            resets_at: Some(1_786_150_800),
        }),
        saved_path: None,
        input_images: Vec::new(),
        imagegen_request_id: None,
        generation_id: None,
    };
    let image_completion =
        RolloutItem::EventMsg(EventMsg::ImageGenerationEnd(ImageGenerationEndEvent {
            call_id: expected_item.id.clone(),
            status: expected_item.status.clone(),
            revised_prompt: expected_item.revised_prompt.clone(),
            image: expected_item.image.clone(),
            transparent_background: expected_item.transparent_background,
            failure: expected_item.failure.clone(),
            saved_path: expected_item.saved_path.clone(),
            input_images: Vec::new(),
            generation_id: expected_item.generation_id.clone(),
        }));
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("image-turn"),
            image_completion,
            completed("image-turn"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy image completion");

    let migrated_item = read_rollout(&path)
        .into_iter()
        .find_map(|line| match line.item {
            RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => Some(event.item),
            _ => None,
        })
        .expect("migrated image completion");
    let TurnItem::Extension(ExtensionItem::ImageGeneration(migrated_item)) = migrated_item else {
        panic!("expected migrated extension image-generation item");
    };
    assert_eq!(migrated_item, expected_item);
}

#[tokio::test]
async fn migration_preserves_image_generation_input_observations() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let expected_item = ImageGenerationItem {
        id: "image-call".to_string(),
        status: "completed".to_string(),
        revised_prompt: Some("paint a blue whale".to_string()),
        image: GeneratedImageRef::Inline {
            result: "cG5n".to_string(),
        },
        transparent_background: None,
        failure: None,
        saved_path: None,
        input_images: vec![ImageGenerationInputImage {
            path: AbsolutePathBuf::from_absolute_path(home.path().join("reference.png"))
                .expect("absolute input path"),
            sha256: "ab".repeat(32),
        }],
        imagegen_request_id: None,
        generation_id: Some("gen-image-123".to_string()),
    };
    let image_completion =
        RolloutItem::EventMsg(EventMsg::ImageGenerationEnd(ImageGenerationEndEvent {
            call_id: expected_item.id.clone(),
            status: expected_item.status.clone(),
            revised_prompt: expected_item.revised_prompt.clone(),
            image: expected_item.image.clone(),
            transparent_background: expected_item.transparent_background,
            failure: expected_item.failure.clone(),
            saved_path: expected_item.saved_path.clone(),
            input_images: expected_item.input_images.clone(),
            generation_id: expected_item.generation_id.clone(),
        }));
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("image-turn"),
            image_completion,
            completed("image-turn"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy image completion");

    let migrated_item = read_rollout(&path)
        .into_iter()
        .find_map(|line| match line.item {
            RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => Some(event.item),
            _ => None,
        })
        .expect("migrated image completion");
    let TurnItem::Extension(ExtensionItem::ImageGeneration(migrated_item)) = migrated_item else {
        panic!("expected migrated extension image-generation item");
    };
    assert_eq!(migrated_item, expected_item);
}

#[tokio::test]
async fn migration_keeps_late_completions_in_their_original_turn() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("old"),
            user_message("old question"),
            started("current"),
            user_message("current question"),
            exec_completion("old", "call-old", "ok"),
            serde_json::from_value(json!({
                "type": "event_msg",
                "payload": {
                    "type": "mcp_tool_call_end",
                    "call_id": "mcp-old",
                    "turn_id": "old",
                    "invocation": {"server": "slack", "tool": "search", "arguments": {}},
                    "duration": {"secs": 0, "nanos": 0},
                    "result": {"Ok": {"content": []}}
                }
            }))
            .expect("build late legacy MCP completion"),
            agent_message("current answer"),
            completed("current"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate late completion");

    let started_turn_ids = read_rollout(&path)
        .into_iter()
        .filter_map(|line| match line.item {
            RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.turn_id),
            _ => None,
        })
        .collect::<Vec<_>>();
    assert_eq!(started_turn_ids, vec!["old", "current"]);

    let items = store
        .list_items(ListItemsParams {
            item_ids: None,
            thread_id,
            turn_id: None,
            include_archived: false,
            position: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            sort_key: ItemSortKey::CreatedAtOrdinal,
            after_updated_at_ordinal: None,
            before_updated_at_ordinal: None,
        })
        .await
        .expect("read projected late completion");
    let item_turn_ids = items
        .items
        .iter()
        .map(|item| item.turn_id.as_str())
        .collect::<Vec<_>>();
    assert_eq!(
        item_turn_ids,
        vec!["old", "current", "old", "old", "current"]
    );
}

#[tokio::test]
async fn migration_hoists_delayed_session_meta_before_paginated_history() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(home.path(), thread_id, SessionSource::Cli, Vec::new());
    let existing = fs::read_to_string(&path).expect("read legacy rollout");
    let pre_header = RolloutLine {
        timestamp: TIMESTAMP.to_string(),
        ordinal: None,
        item: user_message("before metadata"),
    };
    fs::write(
        &path,
        format!(
            "{}\n{existing}",
            serde_json::to_string(&pre_header).expect("serialize pre-header record")
        ),
    )
    .expect("write pre-header rollout");
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate delayed session metadata");

    let lines = read_rollout(&path);
    assert_eq!(lines[0].ordinal, Some(0));
    assert!(matches!(
        &lines[0].item,
        RolloutItem::SessionMeta(metadata)
            if metadata.meta.id == thread_id
                && metadata.meta.history_mode == ThreadHistoryMode::Paginated
    ));
    let turns = store
        .list_turns(ListTurnsParams {
            thread_id,
            include_archived: false,
            cursor: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            items_view: StoredTurnItemsView::NotLoaded,
        })
        .await
        .expect("read migrated turns");
    assert_eq!(turns.turns.len(), 1);

    let second = store
        .migrate_rollouts(apply_options())
        .await
        .expect("rerun migration");
    assert_eq!(
        second.outcomes[0].status,
        RolloutMigrationStatus::AlreadyPaginated
    );
}

#[tokio::test]
async fn migration_preserves_valid_final_record_without_newline() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("question")],
    );
    let final_line = RolloutLine {
        timestamp: TIMESTAMP.to_string(),
        ordinal: None,
        item: agent_message("answer"),
    };
    fs::OpenOptions::new()
        .append(true)
        .open(&path)
        .expect("open legacy rollout")
        .write_all(
            serde_json::to_string(&final_line)
                .expect("serialize final record")
                .as_bytes(),
        )
        .expect("append final record");
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy rollout");

    assert_eq!(
        read_rollout(&path)
            .iter()
            .filter(|line| matches!(line.item, RolloutItem::EventMsg(EventMsg::ItemCompleted(_))))
            .count(),
        2
    );
}

#[tokio::test]
async fn migration_applies_historical_rollbacks_before_sqlite_projection() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("keep"),
            user_message("keep question"),
            agent_message("keep answer"),
            completed("keep"),
            started("remove"),
            user_message("remove question"),
            agent_message("remove answer"),
            completed("remove"),
            started("shell"),
            completed("shell"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            started("replacement"),
            user_message("replacement question"),
            agent_message("replacement answer"),
            completed("replacement"),
        ],
    );
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate rolled-back thread");
    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);

    let lines = read_rollout(&path);
    assert!(!lines.iter().any(|line| matches!(
        line.item,
        RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_))
    )));
    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(
        turns
            .turns
            .iter()
            .map(|turn| turn.turn_id.as_str())
            .collect::<Vec<_>>(),
        vec!["keep", "replacement"]
    );
}

#[tokio::test]
async fn migration_rolls_back_response_and_inter_agent_user_boundaries() {
    let home = TempDir::new().expect("create Codex home");
    let response_thread_id = ThreadId::new();
    let response_path = write_rollout(
        home.path(),
        response_thread_id,
        SessionSource::Cli,
        vec![
            rollout_response_item(input_response_message("user", "remove response boundary")),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let communication_thread_id = ThreadId::new();
    let communication_path = write_rollout(
        home.path(),
        communication_thread_id,
        SessionSource::Cli,
        vec![
            RolloutItem::InterAgentCommunication(InterAgentCommunication::new(
                AgentPath::root(),
                AgentPath::root().join("worker").expect("worker path"),
                Vec::new(),
                "remove communication boundary".to_string(),
                /*trigger_turn*/ true,
            )),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let contextual_thread_id = ThreadId::new();
    let contextual_path = write_rollout(
        home.path(),
        contextual_thread_id,
        SessionSource::Cli,
        vec![
            rollout_response_item(input_response_message("user", "keep first boundary")),
            rollout_response_item(input_response_message(
                "developer",
                "<managed_developer_instructions>context only</managed_developer_instructions>",
            )),
            rollout_response_item(input_response_message(
                "developer",
                "<permissions instructions>context only</permissions instructions>",
            )),
            rollout_response_item(input_response_message(
                "user",
                "<environment_context>context only</environment_context>",
            )),
            rollout_response_item(input_response_message("user", "remove real user boundary")),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy rollback boundaries");

    assert!(
        !read_rollout(&response_path)
            .iter()
            .any(|line| matches!(line.item, RolloutItem::ResponseItem(_)))
    );
    assert!(
        !read_rollout(&communication_path)
            .iter()
            .any(|line| matches!(line.item, RolloutItem::InterAgentCommunication(_)))
    );
    assert_eq!(
        read_rollout(&contextual_path)
            .into_iter()
            .filter_map(|line| match line.item {
                RolloutItem::ResponseItem(response) => Some(response.into_item()),
                _ => None,
            })
            .collect::<Vec<_>>(),
        vec![input_response_message("user", "keep first boundary")]
    );
}

#[tokio::test]
async fn migration_drops_trailing_context_when_rollback_arrives_before_next_turn() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            rollout_response_item(input_response_message("user", "keep question")),
            rollout_response_item(input_response_message("user", "remove question")),
            rollout_response_item(input_response_message(
                "user",
                "<turn_aborted>remove this context too</turn_aborted>",
            )),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate trailing rollback context");

    assert_eq!(
        read_rollout(&path)
            .into_iter()
            .filter_map(|line| match line.item {
                RolloutItem::ResponseItem(response) => Some(response.into_item()),
                _ => None,
            })
            .collect::<Vec<_>>(),
        vec![input_response_message("user", "keep question")]
    );
}

#[tokio::test]
async fn migration_coalesces_response_first_user_message_rollback_boundary() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let file_id = "file_123".to_string();
    let audio = AudioReference::File {
        file_id: "file_audio".into(),
    };
    let response = ResponseItem::Message {
        status: None,
        encrypted_content: None,
        id: None,
        role: "user".to_string(),
        content: vec![
            ContentItem::InputText {
                text: "remove question".to_string(),
            },
            ContentItem::InputImage {
                image: ImageReference::File {
                    file_id: file_id.clone(),
                },
                detail: None,
            },
            ContentItem::InputAudio {
                audio: audio.clone(),
            },
        ],
        phase: None,
        internal_chat_message_metadata_passthrough: None,
    };
    let event = UserMessageEvent {
        message: "remove question".to_string(),
        file_ids: Some(vec![file_id]),
        audio_references: Some(vec![audio]),
        audio_order: vec![UserMessageAudioKind::File],
        ..Default::default()
    };
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            rollout_response_item(response),
            RolloutItem::EventMsg(EventMsg::UserMessage(event)),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate response-first rollback");

    assert!(
        !read_rollout(&path)
            .iter()
            .any(|line| matches!(line.item, RolloutItem::ResponseItem(_)))
    );
}

#[tokio::test]
async fn migration_does_not_coalesce_distinct_adjacent_user_records() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            rollout_response_item(input_response_message("user", "copied parent question")),
            user_message("child question"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate distinct adjacent user records");

    assert_eq!(
        read_rollout(&path)
            .into_iter()
            .filter_map(|line| match line.item {
                RolloutItem::ResponseItem(response) => Some(response.into_item()),
                _ => None,
            })
            .collect::<Vec<_>>(),
        vec![input_response_message("user", "copied parent question")]
    );
}

#[tokio::test]
async fn migration_keeps_late_completions_for_surviving_turns_across_rollback() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("old"),
            user_message("old question"),
            started("remove"),
            user_message("remove question"),
            completed("old"),
            exec_completion("old", "call-old", "ok"),
            item_completed("old", "reason-old"),
            exec_completion("remove", "call-remove", "ok"),
            completed("remove"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            started("replacement"),
            user_message("replacement question"),
            completed("replacement"),
        ],
    );
    let legacy_turns = build_turns_from_rollout_items(
        &read_rollout(&path)
            .into_iter()
            .map(|line| line.item)
            .collect::<Vec<_>>(),
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate late completion rollback");

    let lines = read_rollout(&path);
    assert!(!lines.iter().any(|line| matches!(
        &line.item,
        RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
            if event.item.id() == "call-remove"
    )));
    assert!(lines.iter().any(|line| matches!(
        &line.item,
        RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
            if event.turn_id == "old" && event.item.id() == "call-old"
    )));
    assert!(lines.iter().any(|line| matches!(
        &line.item,
        RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
            if event.turn_id == "old" && event.item.id() == "reason-old"
    )));
    assert!(lines.iter().any(|line| matches!(
        &line.item,
        RolloutItem::EventMsg(EventMsg::TurnComplete(event)) if event.turn_id == "old"
    )));
    assert!(!lines.iter().any(|line| matches!(
        &line.item,
        RolloutItem::EventMsg(EventMsg::TurnComplete(event)) if event.turn_id == "remove"
    )));
    let turns = store
        .list_turns(ListTurnsParams {
            thread_id,
            include_archived: false,
            cursor: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            items_view: StoredTurnItemsView::Summary,
        })
        .await
        .expect("read late-completion rollback turns");
    let turn_ids = turns
        .turns
        .iter()
        .map(|turn| turn.turn_id.as_str())
        .collect::<Vec<_>>();
    assert_eq!(
        turn_ids,
        legacy_turns
            .iter()
            .map(|turn| turn.id.as_str())
            .collect::<Vec<_>>()
    );
    let items = store
        .list_items(ListItemsParams {
            item_ids: None,
            thread_id,
            turn_id: None,
            include_archived: false,
            position: None,
            page_size: 20,
            sort_direction: SortDirection::Asc,
            sort_key: ItemSortKey::CreatedAtOrdinal,
            after_updated_at_ordinal: None,
            before_updated_at_ordinal: None,
        })
        .await
        .expect("read late-completion rollback items");
    for legacy_turn in &legacy_turns {
        for legacy_item in &legacy_turn.items {
            if legacy_item.id().starts_with("item-") {
                continue;
            }
            assert_eq!(
                items
                    .items
                    .iter()
                    .find(|item| item.item_id == legacy_item.id())
                    .map(|item| item.turn_id.as_str()),
                Some(legacy_turn.id.as_str())
            );
        }
    }
}

#[tokio::test]
async fn migration_rolls_back_inter_agent_metadata_with_its_delivery() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let delivery = InterAgentCommunication::new(
        AgentPath::root(),
        AgentPath::root().join("worker").expect("worker path"),
        Vec::new(),
        "remove delivery".to_string(),
        /*trigger_turn*/ true,
    );
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            RolloutItem::InterAgentCommunicationMetadata { trigger_turn: true },
            rollout_response_item(delivery.to_model_input_item()),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate rolled-back inter-agent delivery");

    assert!(!read_rollout(&path).iter().any(|line| match &line.item {
        RolloutItem::InterAgentCommunicationMetadata { .. } => true,
        RolloutItem::ResponseItem(response_item) => {
            matches!(&response_item.item, ResponseItem::AgentMessage { .. })
        }
        _ => false,
    }));
}

#[tokio::test]
async fn migration_rolls_back_pre_compaction_turns_from_sqlite_history() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let RolloutItem::Compacted(mut checkpoint) = compacted(vec![
        input_response_message("user", "old question"),
        ResponseItem::Message {
            status: None,
            encrypted_content: None,
            id: None,
            role: "assistant".to_string(),
            content: vec![ContentItem::OutputText {
                annotations: None,
                logprobs: None,
                text: "old answer".to_string(),
            }],
            phase: None,
            internal_chat_message_metadata_passthrough: None,
        },
    ]) else {
        unreachable!("compacted helper always creates a compaction checkpoint");
    };
    checkpoint.mcp_resource_origins = Some(McpResourceOriginCheckpoint {
        origins: vec![McpResourceOrigin {
            call_id: "widget-call".to_string(),
            turn_id: Some("keep-before-compaction".to_string()),
            tool: "_product_search".to_string(),
            connector_id: "shopping".to_string(),
            link_id: None,
            uri: "ui://shopping/widget".to_string(),
            ambiguous_account: false,
        }],
        turns: vec!["keep-before-compaction".to_string()],
        current_turn_id: Some("keep-before-compaction".to_string()),
    });
    let answer_event = serde_json::from_value(serde_json::json!({
        "type": "verified_answer", "turn_id": "keep-before-compaction", "call_id": "ask-1",
        "questions": [{"question": "Publish?", "answer": "Only privately."}]
    }))
    .expect("verified answer fixture");
    let wake_answer_event = serde_json::from_value(serde_json::json!({
        "type": "verified_answer", "turn_id": "empty-answer-turn", "call_id": "ask-2",
        "questions": [{"question": "Continue?", "answer": "Keep it private."}]
    }))
    .expect("empty-turn answer fixture");
    checkpoint.retained_context = Some(Default::default());
    let context = checkpoint
        .retained_context
        .as_mut()
        .expect("retained context");
    context.record(&answer_event, &uuid_v7_item_id_generator_for_test());
    context.record(&wake_answer_event, &uuid_v7_item_id_generator_for_test());
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("keep-before-compaction"),
            user_message("old question"),
            RolloutItem::RetainedContext(answer_event),
            completed("keep-before-compaction"),
            started("empty-answer-turn"),
            RolloutItem::RetainedContext(wake_answer_event),
            completed("empty-answer-turn"),
            RolloutItem::Compacted(checkpoint),
            started("remove-after-compaction"),
            user_message("new question"),
            completed("remove-after-compaction"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 2,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate rollback through compaction");

    let turns = store
        .list_turns(ListTurnsParams {
            thread_id,
            include_archived: false,
            cursor: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            items_view: StoredTurnItemsView::Summary,
        })
        .await
        .expect("read rollback-through-compaction turns");
    assert_eq!(turns.turns.len(), 1);
    let migrated = read_rollout(&path);
    assert!(
        !migrated
            .iter()
            .any(|line| matches!(line.item, RolloutItem::RetainedContext(_)))
    );
    let checkpoint = migrated
        .into_iter()
        .find_map(|line| match line.item {
            RolloutItem::Compacted(item) => Some(item),
            _ => None,
        })
        .expect("retained compaction");
    assert_eq!(checkpoint.replacement_history, Some(Vec::new()));
    assert_eq!(checkpoint.mcp_resource_origins, None);
    assert_eq!(
        checkpoint
            .retained_context
            .expect("retained context checkpoint")
            .verified_answers()
            .count(),
        0
    );
}

#[tokio::test]
async fn migration_preserves_answers_before_a_rolled_back_steer() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let mut items = vec![started("shared-turn")];
    let mut answers = Vec::new();
    let RolloutItem::Compacted(mut checkpoint) = compacted(vec![
        input_response_message("user", "before-steer"),
        input_response_message("user", "after-steer"),
    ]) else {
        unreachable!()
    };
    checkpoint.retained_context = Some(Default::default());
    for call_id in ["before-steer", "after-steer"] {
        items.push(user_message(call_id));
        let mut call: ResponseItem = serde_json::from_value(json!({
            "type": "function_call", "call_id": call_id,
            "name": "request_user_input", "arguments": "{}"
        }))
        .expect("request_user_input call");
        call.set_turn_id_if_missing("shared-turn");
        items.push(rollout_response_item(call));
        let answer: codex_rollout::RetainedContextEvent = serde_json::from_value(json!({
            "type": "verified_answer", "turn_id": "shared-turn", "call_id": call_id,
            "questions": [{"question": "Publish?", "answer": "Only privately."}]
        }))
        .expect("verified answer");
        checkpoint
            .retained_context
            .as_mut()
            .expect("retained context")
            .record(&answer, &uuid_v7_item_id_generator_for_test());
        answers.push(answer);
    }
    // Delayed delivery must use the original call boundary, not the turn's latest steer.
    items.extend(answers.iter().cloned().map(RolloutItem::RetainedContext));
    items.push(completed("shared-turn"));
    items.push(RolloutItem::Compacted(checkpoint));
    items.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
        ThreadRolledBackEvent { num_turns: 1 },
    )));
    let path = write_rollout(home.path(), thread_id, SessionSource::Cli, items);
    let store = indexed_store(home.path()).await;
    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate rollback of a steer");
    let migrated = read_rollout(&path);
    let events = migrated
        .iter()
        .filter_map(|line| match &line.item {
            RolloutItem::RetainedContext(event) => Some(event.clone()),
            _ => None,
        })
        .collect::<Vec<_>>();
    assert_eq!(events, answers[..1]);
    let checkpoint = migrated
        .iter()
        .find_map(|line| match &line.item {
            RolloutItem::Compacted(checkpoint) => checkpoint.retained_context.as_ref(),
            _ => None,
        })
        .expect("retained checkpoint");
    assert_eq!(
        checkpoint
            .verified_answers()
            .cloned()
            .map(
                |answer| codex_rollout::RetainedContextEvent::VerifiedAnswer {
                    answer,
                    acceptance_order: None,
                }
            )
            .collect::<Vec<_>>(),
        answers[..1]
    );
}

fn delivery(turn_id: &str, message_id: &str, text: &str, order: u64) -> RetainedContextEvent {
    serde_json::from_value(json!({
        "type": "delivered_assistant_message", "turn_id": turn_id,
        "message_id": message_id, "text": text, "complete": true,
        "acceptance_order": order
    }))
    .expect("nested delivery")
}

fn ordered_response(item: ResponseItem, order: u64) -> RolloutItem {
    let mut envelope = codex_rollout::ResponseItemEnvelope::new(item);
    envelope.metadata.get_or_insert_default().user_input_order = Some(order);
    RolloutItem::ResponseItem(envelope)
}

fn worker_communication(text: &str) -> InterAgentCommunication {
    InterAgentCommunication::new(
        AgentPath::root(),
        AgentPath::root().join("worker").expect("worker path"),
        Vec::new(),
        text.to_owned(),
        /*trigger_turn*/ true,
    )
}

async fn migrate_after_rollback(mut items: Vec<RolloutItem>) -> Vec<RolloutItem> {
    items.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
        ThreadRolledBackEvent { num_turns: 1 },
    )));
    let home = TempDir::new().expect("create Codex home");
    let path = write_rollout(home.path(), ThreadId::new(), SessionSource::Cli, items);
    indexed_store(home.path())
        .await
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate rollout after rollback");
    read_rollout(&path)
        .into_iter()
        .map(|line| line.item)
        .collect()
}

fn retained_deliveries(items: &[RolloutItem]) -> Vec<&RetainedContextEvent> {
    items
        .iter()
        .filter_map(|item| match item {
            RolloutItem::RetainedContext(
                event @ RetainedContextEvent::DeliveredAssistantMessage { .. },
            ) => Some(event),
            _ => None,
        })
        .collect()
}

#[tokio::test]
async fn migration_rolls_back_nested_deliveries_by_acceptance_order() {
    let [initial, steer] = ["initial", "steer"].map(|text| {
        let mut item = input_response_message("user", text);
        item.set_turn_id_if_missing("shared-turn");
        item
    });
    let deliveries = [("exec:0", "first", 1), ("exec:1", "second", 3)]
        .map(|(message_id, text, order)| delivery("shared-turn", message_id, text, order));
    let RolloutItem::Compacted(mut checkpoint) = compacted(vec![initial.clone(), steer.clone()])
    else {
        unreachable!("compacted helper creates a checkpoint");
    };
    let context = checkpoint.retained_context.insert(Default::default());
    for (text, order) in [("initial", 0), ("steer", 2)] {
        context.record_user_message(
            serde_json::from_value(json!({
                "turn_id": "shared-turn", "message_id": text, "text": text, "complete": true
            }))
            .expect("retained instruction"),
            codex_rollout::RetainedInputSource::Local(Some(order)),
            &uuid_v7_item_id_generator_for_test(),
        );
    }
    context.record(&deliveries[0], &uuid_v7_item_id_generator_for_test());
    let first_retained_message =
        serde_json::to_value(&*context).expect("retained context")["assistant_messages"][0].clone();
    context.record(&deliveries[1], &uuid_v7_item_id_generator_for_test());
    let outer_call = serde_json::from_value(json!({
        "type": "function_call", "call_id": "exec", "name": "exec", "arguments": "{}"
    }))
    .expect("outer code-mode call");
    let migrated = migrate_after_rollback(vec![
        started("shared-turn"),
        ordered_response(initial, /*order*/ 0),
        user_message("initial"),
        rollout_response_item(outer_call),
        RolloutItem::RetainedContext(deliveries[1].clone()),
        ordered_response(steer, /*order*/ 2),
        user_message("steer"),
        RolloutItem::RetainedContext(deliveries[0].clone()),
        completed("shared-turn"),
        started("compaction-turn"),
        RolloutItem::Compacted(checkpoint),
        completed("compaction-turn"),
    ])
    .await;
    let retained = migrated.iter().find_map(|item| match item {
        RolloutItem::Compacted(checkpoint) => checkpoint.retained_context.as_ref(),
        _ => None,
    });
    assert_eq!(retained_deliveries(&migrated), [&deliveries[0]]);
    assert_eq!(
        serde_json::to_value(retained.expect("retained checkpoint")).expect("retained context")["assistant_messages"],
        json!([first_retained_message])
    );
}

#[tokio::test]
async fn migration_keeps_delivery_before_newer_unsequenced_communication() {
    let mut initial = input_response_message("user", "initial");
    initial.set_turn_id_if_missing("root-turn");
    let delivery = delivery("root-turn", "exec:0", "May I publish?", /*order*/ 1);
    let migrated = migrate_after_rollback(vec![
        started("root-turn"),
        ordered_response(initial, /*order*/ 0),
        user_message("initial"),
        RolloutItem::InterAgentCommunication(worker_communication("later communication")),
        RolloutItem::RetainedContext(delivery.clone()),
    ])
    .await;
    assert_eq!(retained_deliveries(&migrated), [&delivery]);
}

#[tokio::test]
async fn migration_removes_delivery_from_rolled_back_communication_turn() {
    for encoding in [
        "standalone",
        "unsequenced pair",
        "sequenced pair",
        "sequenced same turn",
    ] {
        let communication = worker_communication("start communication turn");
        let mut items = vec![
            started("root-turn"),
            ordered_response(input_response_message("user", "initial"), /*order*/ 0),
            user_message("initial"),
        ];
        let delivery_turn = if encoding == "sequenced same turn" {
            "root-turn"
        } else {
            items.extend([completed("root-turn"), started("communication-turn")]);
            "communication-turn"
        };
        if encoding == "standalone" {
            items.push(RolloutItem::InterAgentCommunication(communication));
        } else {
            let response = communication.to_model_input_item();
            let response = if encoding.starts_with("sequenced") {
                ordered_response(response, /*order*/ 1)
            } else {
                rollout_response_item(response)
            };
            items.extend([
                RolloutItem::InterAgentCommunicationMetadata { trigger_turn: true },
                response,
            ]);
        }
        let delivery = delivery(delivery_turn, "exec:0", "May I publish?", /*order*/ 2);
        items.extend([
            RolloutItem::RetainedContext(delivery),
            completed(delivery_turn),
        ]);
        assert!(
            retained_deliveries(&migrate_after_rollback(items).await).is_empty(),
            "{encoding}"
        );
    }
}

#[tokio::test]
async fn migration_removes_delivery_with_only_legacy_unsequenced_instruction() {
    let delivery = delivery("legacy-turn", "exec:0", "May I publish?", /*order*/ 1);
    let migrated = migrate_after_rollback(vec![
        started("legacy-turn"),
        rollout_response_item(input_response_message("user", "legacy instruction")),
        user_message("legacy instruction"),
        RolloutItem::RetainedContext(delivery),
    ])
    .await;
    assert!(retained_deliveries(&migrated).is_empty());
}

#[tokio::test]
async fn migration_keeps_downgraded_delivery_before_later_same_turn_steer() {
    let delivery = delivery("shared-turn", "exec:0", "May I publish?", /*order*/ 1);
    let mut legacy_wire = serde_json::to_value(RolloutItem::RetainedContext(delivery))
        .expect("compatible response-item encoding");
    // Older readers preserve the assistant response but drop this new metadata field.
    legacy_wire["metadata"]
        .as_object_mut()
        .expect("response-item metadata")
        .remove("delivered_assistant_message");
    let old_read: RolloutItem = serde_json::from_value(legacy_wire).expect("older response item");
    assert!(matches!(old_read, RolloutItem::ResponseItem(_)));
    let [initial, steer] = ["initial", "steer"].map(|text| {
        let mut item = input_response_message("user", text);
        item.set_turn_id_if_missing("shared-turn");
        item
    });
    let migrated = migrate_after_rollback(vec![
        started("shared-turn"),
        ordered_response(initial, /*order*/ 0),
        user_message("initial"),
        old_read,
        ordered_response(steer, /*order*/ 2),
        user_message("steer"),
    ])
    .await;
    assert!(migrated.into_iter().any(|item| matches!(item,
        RolloutItem::ResponseItem(envelope)
            if matches!(&envelope.item, ResponseItem::Message { role, content, .. }
                if role == "assistant"
                    && matches!(content.as_slice(), [ContentItem::OutputText { text , .. }]
                        if text == "May I publish?"))
    )));
}

#[tokio::test]
async fn migration_preserves_reverse_replay_anchor_after_pre_compaction_rollback() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            rollout_response_item(input_response_message("user", "remove question")),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            compacted(vec![input_response_message("user", "old question")]),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate pre-compaction rollback");

    let replacement_history = read_rollout(&path)
        .into_iter()
        .find_map(|line| match line.item {
            RolloutItem::Compacted(item) => item.replacement_history,
            _ => None,
        })
        .expect("retained compaction");
    assert!(replacement_history.is_empty());
}

#[tokio::test]
async fn migration_preserves_same_turn_evidence_across_rollback_checkpoints() {
    assert_migrated_evidence_order(/*steer_order*/ None).await;
}

#[tokio::test]
async fn migration_removes_answers_accepted_after_a_queued_steer() {
    assert_migrated_evidence_order(Some(1)).await;
}

async fn assert_migrated_evidence_order(steer_order: Option<u64>) {
    const INITIAL: &str = "Never publish publicly.";
    const STEER: &str = "Also inspect the README.";
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let [initial, steer] =
        [("initial", INITIAL), ("steer", STEER)].map(|(id, text)| ResponseItem::Message {
            status: None,
            encrypted_content: None,
            id: Some(ResponseItemId::with_suffix("msg", id)),
            role: "user".to_owned(),
            content: vec![ContentItem::InputText {
                text: text.to_owned(),
            }],
            phase: None,
            internal_chat_message_metadata_passthrough: Some(
                InternalChatMessageMetadataPassthrough {
                    turn_id: Some("shared-turn".to_owned()),
                    content_item_kinds: Some(vec![ContentItemKind("user.text".to_owned())]),
                    ..Default::default()
                },
            ),
        });
    let RolloutItem::Compacted(mut checkpoint) = compacted(vec![initial.clone(), steer.clone()])
    else {
        unreachable!("compacted helper creates a checkpoint");
    };
    let retained = json!({
        "user_messages": [
            {"order": 0, "turn_id": "shared-turn", "message_id": "msg_initial", "text": INITIAL, "complete": true},
            {"order": steer_order.unwrap_or(2), "turn_id": "shared-turn", "message_id": "msg_steer", "text": STEER, "complete": true}
        ],
        "verified_answers": [
            {"order": if steer_order.is_some() { 2 } else { 1 }, "turn_id": "shared-turn", "call_id": "before", "questions": [{"question": "Publish?", "answer": "Only privately."}]},
            {"order": 3, "turn_id": "shared-turn", "call_id": "after", "questions": [{"question": "Publish the README?", "answer": "Do not publish it."}]}
        ],
        "incomplete": false, "user_messages_incomplete": false, "next_order": 4,
        "assistant_messages": [], "assistant_messages_incomplete": true
    });
    checkpoint.retained_context =
        Some(serde_json::from_value(retained.clone()).expect("retained fixture"));
    let answers = checkpoint
        .retained_context
        .as_ref()
        .expect("retained checkpoint")
        .verified_answers()
        .cloned()
        .map(|answer| {
            let acceptance_order =
                steer_order.map(|_| if answer.call_id == "before" { 2 } else { 3 });
            codex_rollout::RetainedContextEvent::VerifiedAnswer {
                answer,
                acceptance_order,
            }
        })
        .collect::<Vec<_>>();
    let mut before_retained = retained.clone();
    before_retained["user_messages"]
        .as_array_mut()
        .expect("user messages")
        .truncate(/*len*/ 1);
    before_retained["verified_answers"]
        .as_array_mut()
        .expect("verified answers")
        .truncate(/*len*/ 1);
    before_retained["next_order"] = json!(if steer_order.is_some() { 3 } else { 2 });
    let remaining_answers = usize::from(steer_order.is_none());
    let mut expected = retained;
    expected["user_messages"]
        .as_array_mut()
        .expect("user messages")
        .truncate(/*len*/ 1);
    expected["verified_answers"]
        .as_array_mut()
        .expect("verified answers")
        .truncate(remaining_answers);
    let mut before_expected = expected.clone();
    before_expected["next_order"] = before_retained["next_order"].clone();
    let mut before_steer = checkpoint.clone();
    before_steer.replacement_history = Some(vec![initial.clone().into()]);
    before_steer.retained_context =
        Some(serde_json::from_value(before_retained).expect("pre-steer checkpoint"));
    let mut after_rollback = before_steer.clone();
    after_rollback.retained_context =
        Some(serde_json::from_value(expected.clone()).expect("post-rollback checkpoint"));
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("shared-turn"),
            rollout_response_item(initial.clone()),
            user_message(INITIAL),
            RolloutItem::RetainedContext(answers[0].clone()),
            RolloutItem::ResponseItem(codex_rollout::ResponseItemEnvelope {
                item: serde_json::from_value(json!({
                    "type": "message", "id": "late-assistant", "role": "assistant",
                    "content": [{"type": "output_text", "text": "Late assistant question?"}],
                    "internal_chat_message_metadata_passthrough": {"turn_id": "shared-turn"}
                }))
                .expect("assistant message"),
                metadata: Some(
                    serde_json::from_value(json!({"user_input_order": 2}))
                        .expect("assistant order"),
                ),
            }),
            RolloutItem::Compacted(before_steer),
            RolloutItem::ResponseItem(codex_rollout::ResponseItemEnvelope {
                item: steer,
                metadata: steer_order.map(|order| {
                    serde_json::from_value(json!({
                        "user_input_order": order
                    }))
                    .expect("acceptance metadata")
                }),
            }),
            user_message(STEER),
            RolloutItem::RetainedContext(answers[1].clone()),
            completed("shared-turn"),
            started("compaction-turn"),
            RolloutItem::Compacted(checkpoint),
            completed("compaction-turn"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            // This checkpoint already contains the correct live rollback result.
            started("post-rollback-compaction"),
            RolloutItem::Compacted(after_rollback),
            completed("post-rollback-compaction"),
        ],
    );
    let store = indexed_store(home.path()).await;
    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate same-turn rollback");
    let migrated = read_rollout(&path);
    assert_eq!(
        migrated.iter().any(|line| matches!(&line.item,
            RolloutItem::ResponseItem(envelope)
                if envelope.item.id().is_some_and(|id| id.as_str() == "late-assistant")
        )),
        steer_order.is_none()
    );
    let checkpoints = migrated
        .iter()
        .filter_map(|line| match &line.item {
            RolloutItem::Compacted(item) => Some(item),
            _ => None,
        })
        .collect::<Vec<_>>();
    assert_eq!(
        checkpoints
            .iter()
            .map(
                |checkpoint| serde_json::to_value(&checkpoint.retained_context)
                    .expect("retained context")
            )
            .collect::<Vec<_>>(),
        vec![before_expected, expected.clone(), expected]
    );
    assert_eq!(
        checkpoints
            .last()
            .expect("latest checkpoint")
            .replacement_history,
        Some(vec![initial.into()])
    );
    assert_eq!(
        migrated
            .into_iter()
            .filter_map(|line| match line.item {
                RolloutItem::RetainedContext(event) => Some(event),
                _ => None,
            })
            .collect::<Vec<_>>(),
        answers[..remaining_answers]
    );
}

#[tokio::test]
async fn migration_keeps_empty_replay_anchor_from_rolled_back_turn() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("keep"),
            user_message("keep question"),
            completed("keep"),
            started("remove"),
            user_message("remove question"),
            compacted(vec![input_response_message("user", "old question")]),
            completed("remove"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            started("replacement"),
            user_message("replacement question"),
            completed("replacement"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate rolled-back replay anchor");

    let replacement_history = read_rollout(&path)
        .into_iter()
        .find_map(|line| match line.item {
            RolloutItem::Compacted(item) => item.replacement_history,
            _ => None,
        })
        .expect("retained compaction");
    assert!(replacement_history.is_empty());
}

#[tokio::test]
async fn migration_uses_turn_context_to_select_reverse_replay_anchor() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("keep"),
            user_message("keep question"),
            compacted(vec![input_response_message("user", "keep question")]),
            completed("keep"),
            started("remove"),
            user_message("remove question"),
            serde_json::from_value(json!({
                "type": "event_msg",
                "payload": {
                    "type": "turn_aborted",
                    "turn_id": "other",
                    "reason": "interrupted"
                }
            }))
            .expect("build mismatched turn abort"),
            serde_json::from_value(json!({
                "type": "turn_context",
                "payload": {
                    "turn_id": "remove",
                    "cwd": home.path(),
                    "approval_policy": "never",
                    "sandbox_policy": {"type": "read-only"},
                    "model": "test-model",
                    "summary": "auto"
                }
            }))
            .expect("build turn context"),
            compacted(vec![input_response_message("user", "remove question")]),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            started("replacement"),
            user_message("replacement question"),
            completed("replacement"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate turn-context replay anchor");

    let replacement_histories = read_rollout(&path)
        .into_iter()
        .filter_map(|line| match line.item {
            RolloutItem::Compacted(item) => item.replacement_history.map(|items| {
                items
                    .into_iter()
                    .map(codex_rollout::ResponseItemEnvelope::into_item)
                    .collect::<Vec<_>>()
            }),
            _ => None,
        })
        .collect::<Vec<_>>();
    assert_eq!(
        replacement_histories,
        vec![vec![input_response_message("user", "keep question")]]
    );
}

#[tokio::test]
async fn migration_applies_cumulative_and_overflowing_rollbacks() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            started("first"),
            user_message("first question"),
            completed("first"),
            started("second"),
            user_message("second question"),
            completed("second"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 1,
            })),
            started("third"),
            user_message("third question"),
            completed("third"),
            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
                num_turns: 99,
            })),
            user_message("replacement question"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate cumulative rollbacks");

    let turns = store
        .list_turns(ListTurnsParams {
            thread_id,
            include_archived: false,
            cursor: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            items_view: StoredTurnItemsView::Summary,
        })
        .await
        .expect("read cumulative rollback turns");
    assert_eq!(turns.turns.len(), 1);
    assert!(!read_rollout(&path).iter().any(|line| matches!(
        line.item,
        RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_))
    )));
}

#[tokio::test]
async fn migration_drops_copied_user_fork_metadata_without_creating_a_history_base() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let parent_id = ThreadId::new();
    let copied_metadata = SessionMeta {
        session_id: parent_id.into(),
        id: parent_id,
        timestamp: TIMESTAMP.to_string(),
        cwd: home.path().to_path_buf(),
        source: SessionSource::Cli,
        ..SessionMeta::default()
    };
    let copied_response =
        rollout_response_item(input_response_message("user", "copied parent history"));
    let path = write_rollout_with_fork(
        home.path(),
        thread_id,
        SessionSource::Cli,
        Some(parent_id),
        vec![
            RolloutItem::SessionMeta(Box::new(SessionMetaLine {
                meta: copied_metadata,
                git: None,
            })),
            copied_response,
            user_message("child question"),
            agent_message("child answer"),
        ],
    );
    let expected_responses = read_rollout(&path)
        .into_iter()
        .filter_map(|line| match line.item {
            RolloutItem::ResponseItem(item) => {
                Some(serde_json::to_value(item.item).expect("serialize copied response"))
            }
            _ => None,
        })
        .collect::<Vec<_>>();
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate copied user fork");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    let lines = read_rollout(&path);
    assert!(matches!(
        &lines[0].item,
        RolloutItem::SessionMeta(metadata)
            if metadata.meta.id == thread_id
                && metadata.meta.forked_from_id == Some(parent_id)
                && metadata.meta.history_mode == ThreadHistoryMode::Paginated
                && metadata.meta.history_base.is_none()
    ));
    assert_eq!(
        lines
            .iter()
            .filter(|line| matches!(line.item, RolloutItem::SessionMeta(_)))
            .count(),
        1
    );
    assert_eq!(
        lines
            .into_iter()
            .filter_map(|line| match line.item {
                RolloutItem::ResponseItem(item) => {
                    Some(serde_json::to_value(item.item).expect("serialize migrated response"))
                }
                _ => None,
            })
            .collect::<Vec<_>>(),
        expected_responses
    );
}

#[tokio::test]
async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::SubAgent(SubAgentSource::Other("test".to_string())),
        vec![
            RolloutItem::Compacted(Box::new(CompactedItem {
                message: "superseded checkpoint".repeat(1024),
                replacement_history: Some(Vec::new()),
                retained_context: None,
                guardian_history: None,
                mcp_resource_origins: None,
                window_number: Some(1),
                first_window_id: None,
                previous_window_id: None,
                window_id: None,
                compaction_response_id: None,
                latest_token_usage_record: None,
                resume_metadata: None,
            })),
            RolloutItem::Compacted(Box::new(CompactedItem {
                message: "latest checkpoint".to_string(),
                replacement_history: Some(vec![
                    ResponseItem::Message {
                        encrypted_content: None,
                        status: None,
                        id: None,
                        role: "user".to_string(),
                        content: vec![ContentItem::InputText {
                            text: "latest compacted context".to_string(),
                        }],
                        phase: None,
                        internal_chat_message_metadata_passthrough: None,
                    }
                    .into(),
                ]),
                retained_context: None,
                guardian_history: None,
                mcp_resource_origins: None,
                window_number: Some(2),
                first_window_id: None,
                previous_window_id: None,
                window_id: None,
                compaction_response_id: None,
                latest_token_usage_record: None,
                resume_metadata: None,
            })),
            started("child-turn"),
            RolloutItem::TurnContext(TurnContextItem {
                turn_id: Some("child-turn".to_string()),
                root_turn_id: None,
                direct_only_tool_namespaces: None,
                disabled_plugin_ids: None,
                cwd: serde_json::from_value(json!(home.path())).expect("absolute cwd"),
                approval_policy: AskForApproval::Never,
                approvals_reviewer: None,
                sandbox_policy: SandboxPolicy::new_read_only_policy(),
                permission_profile: None,
                active_permission_profile: None,
                file_system_sandbox_policy: None,
                model: "test-model".to_string(),
                comp_hash: None,
                collaboration_mode: None,
                multi_agent_version: None,
                realtime_active: None,
                cyber_access_program: None,
                effort: None,
                summary: Some(ReasoningSummary::Auto),
            }),
            user_message("child question"),
            agent_message("child answer"),
            completed("child-turn"),
        ],
    );
    writeln!(
        fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .expect("open legacy subagent rollout"),
        "{{not valid rollout json"
    )
    .expect("append malformed record");
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate legacy subagent");

    let lines = read_rollout(&path);
    let RolloutItem::SessionMeta(metadata) = &lines[0].item else {
        panic!("migrated rollout should start with session metadata");
    };
    assert_eq!(
        metadata.meta.subagent_history_start_ordinal,
        Some(lines.len() as u64)
    );
    assert!(
        !fs::read_to_string(&path)
            .expect("read migrated rollout")
            .contains("superseded checkpoint")
    );
    let context = store
        .load_latest_model_context(LoadThreadHistoryParams {
            thread_id,
            include_archived: false,
        })
        .await
        .expect("load migrated model context");
    assert!(context.items.iter().any(|item| {
        matches!(item, RolloutItem::Compacted(compacted) if compacted.message == "latest checkpoint")
    }));
    assert!(
        list_active_summary_turns(&store, thread_id)
            .await
            .turns
            .is_empty()
    );
}

#[tokio::test]
async fn migration_keeps_small_uncompacted_subagent_replay_as_prefix() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::SubAgent(SubAgentSource::Other("test".to_string())),
        vec![
            user_message("child question"),
            agent_message("child answer"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate uncompacted legacy subagent");

    let lines = read_rollout(&path);
    let RolloutItem::SessionMeta(metadata) = &lines[0].item else {
        panic!("migrated rollout should start with session metadata");
    };
    assert_eq!(
        metadata.meta.subagent_history_start_ordinal,
        Some(lines.len() as u64)
    );
    assert_eq!(
        lines
            .iter()
            .filter(|line| matches!(line.item, RolloutItem::EventMsg(EventMsg::ItemCompleted(_))))
            .count(),
        2
    );
}

#[tokio::test]
async fn migration_projects_memory_consolidation_as_ordinary_history() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::SubAgent(SubAgentSource::MemoryConsolidation),
        vec![
            user_message("memory question"),
            agent_message("memory answer"),
        ],
    );
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate memory consolidation rollout");

    let lines = read_rollout(&path);
    assert!(matches!(
        &lines[0].item,
        RolloutItem::SessionMeta(metadata)
            if metadata.meta.history_mode == ThreadHistoryMode::Paginated
                && metadata.meta.subagent_history_start_ordinal.is_none()
    ));
    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(turns.turns.len(), 1);
    assert_eq!(turns.turns[0].items.len(), 2);
}

#[tokio::test]
async fn dry_run_reports_migration_order() {
    let home = TempDir::new().expect("create Codex home");
    let root_id = ThreadId::new();
    let root = write_rollout(
        home.path(),
        root_id,
        SessionSource::Cli,
        vec![user_message("root question")],
    );
    let newest_directory = home.path().join("sessions/2025/01/04");
    fs::create_dir_all(&newest_directory).expect("create newest rollout directory");
    let newest_root = newest_directory.join(format!("rollout-2025-01-04T12-00-00-{root_id}.jsonl"));
    fs::rename(&root, &newest_root).expect("move root rollout to newest date");
    let root = newest_root;
    let subagent_id = ThreadId::new();
    let subagent = write_rollout(
        home.path(),
        subagent_id,
        SessionSource::SubAgent(SubAgentSource::Other("test".to_string())),
        vec![user_message("subagent question")],
    );
    let memory_id = ThreadId::new();
    let memory = write_rollout(
        home.path(),
        memory_id,
        SessionSource::SubAgent(SubAgentSource::MemoryConsolidation),
        vec![user_message("memory question")],
    );
    let memory_directory = home.path().join("sessions/2025/01/01");
    fs::create_dir_all(&memory_directory).expect("create memory rollout directory");
    let moved_memory = memory_directory.join(memory.file_name().expect("memory filename"));
    fs::rename(memory, &moved_memory).expect("move memory rollout");
    let compressed_id = ThreadId::new();
    let oldest_directory = home.path().join("sessions/2025/01/02");
    fs::create_dir_all(&oldest_directory).expect("create oldest rollout directory");
    let source_compressed_plain = write_rollout(
        home.path(),
        compressed_id,
        SessionSource::Cli,
        vec![user_message("compressed question")],
    );
    let compressed_plain =
        oldest_directory.join(format!("rollout-2025-01-02T12-00-00-{compressed_id}.jsonl"));
    fs::rename(source_compressed_plain, &compressed_plain).expect("move compressed rollout");
    let compressed = compress_rollout(&compressed_plain);
    let archived_id = ThreadId::new();
    let archived = move_to_archived(
        home.path(),
        write_rollout(
            home.path(),
            archived_id,
            SessionSource::Cli,
            vec![user_message("archived question")],
        ),
    );
    let original = fs::read(&root).expect("read original root rollout");
    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);

    let mut progress = Vec::new();
    let report = store
        .migrate_rollouts_with_progress(RolloutMigrationOptions::default(), |update| {
            progress.push(update);
        })
        .await
        .expect("inspect legacy rollouts");

    let expected = vec![
        (root.clone(), root_id, RolloutMigrationStatus::Eligible),
        (subagent, subagent_id, RolloutMigrationStatus::Eligible),
        (compressed, compressed_id, RolloutMigrationStatus::Eligible),
        (moved_memory, memory_id, RolloutMigrationStatus::Eligible),
        (archived, archived_id, RolloutMigrationStatus::Eligible),
    ];
    assert_eq!(
        report
            .outcomes
            .iter()
            .map(|outcome| (
                outcome.rollout_path.clone(),
                outcome.thread_id.expect("rollout thread ID"),
                outcome.status,
            ))
            .collect::<Vec<_>>(),
        expected,
    );
    assert_eq!(
        progress.last(),
        Some(&RolloutMigrationProgress {
            processed_paths: 5,
            total_paths: 5,
            outcome_status: Some(RolloutMigrationStatus::Eligible),
        })
    );
    assert_eq!(fs::read(&root).expect("read inspected rollout"), original);

    let selected = store
        .migrate_rollouts(RolloutMigrationOptions {
            thread_ids: vec![root_id],
            ..RolloutMigrationOptions::default()
        })
        .await
        .expect("inspect selected rollout");
    assert_eq!(selected.outcomes.len(), 1);
    assert_eq!(selected.outcomes[0].thread_id, Some(root_id));
    assert_eq!(
        selected.outcomes[0].status,
        RolloutMigrationStatus::Eligible
    );
}

#[tokio::test]
async fn migration_preserves_compressed_rollouts_during_publish_and_recovery() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            user_message("compressed question"),
            agent_message("compressed answer"),
        ],
    );
    let compressed_path = compress_rollout(&path);
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate compressed rollout");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(!path.exists());
    assert!(compressed_path.exists());
    assert_eq!(
        codex_rollout::read_session_meta_line(&compressed_path)
            .await
            .expect("read compressed metadata")
            .meta
            .history_mode,
        ThreadHistoryMode::Paginated
    );
    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(turns.turns.len(), 1);
    assert_eq!(turns.turns[0].items.len(), 2);

    thread_history::delete_thread(&store, thread_id)
        .await
        .expect("simulate missing projection");
    let journal_path = migration_journal_path(home.path(), thread_id);
    write_migration_journal(&journal_path)
        .await
        .expect("simulate pending migration journal");
    let recovered = store
        .migrate_rollouts(apply_options())
        .await
        .expect("recover compressed rollout");

    assert_eq!(
        recovered.outcomes[0].status,
        RolloutMigrationStatus::Migrated
    );
    assert!(recovered.outcomes[0].bytes_processed > 0);
    assert!(compressed_path.exists());
    assert!(!path.exists());
    assert!(!journal_path.exists());
}

#[tokio::test]
async fn migration_migrates_archived_rollouts_without_unarchiving_them() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let active_path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            user_message("archived question"),
            agent_message("archived answer"),
        ],
    );
    let archived_path = move_to_archived(home.path(), active_path.clone());
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate archived rollout");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(!active_path.exists());
    assert!(archived_path.exists());
    assert!(matches!(
        &read_rollout(&archived_path)[0].item,
        RolloutItem::SessionMeta(metadata)
            if metadata.meta.history_mode == ThreadHistoryMode::Paginated
    ));
    let turns = store
        .list_turns(ListTurnsParams {
            thread_id,
            include_archived: true,
            cursor: None,
            page_size: 10,
            sort_direction: SortDirection::Asc,
            items_view: StoredTurnItemsView::Summary,
        })
        .await
        .expect("read archived projected turns");
    assert_eq!(turns.turns.len(), 1);
    assert_eq!(turns.turns[0].items.len(), 2);
}

#[tokio::test]
async fn migration_retries_a_rollout_moved_after_path_discovery() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let active_path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("question"), agent_message("answer")],
    );
    let store = indexed_store(home.path()).await;
    let archived_path = move_to_archived(home.path(), active_path.clone());

    let report = store
        .migrate_rollouts_with_progress_for_trigger(
            apply_options(),
            |_| {},
            RolloutMigrationTrigger::Startup,
            RolloutMigrationPaths::Known(vec![active_path]),
        )
        .await
        .expect("migrate moved rollout");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(matches!(
        &read_rollout(&archived_path)[0].item,
        RolloutItem::SessionMeta(metadata)
            if metadata.meta.history_mode == ThreadHistoryMode::Paginated
    ));
}

#[tokio::test]
async fn migration_preserves_legacy_displayed_thread_names() {
    let home = TempDir::new().expect("create Codex home");
    let title_thread_id = ThreadId::new();
    write_rollout(
        home.path(),
        title_thread_id,
        SessionSource::SubAgent(SubAgentSource::Other("guardian".to_string())),
        vec![user_message("title question")],
    );
    let index_thread_id = ThreadId::new();
    write_rollout(
        home.path(),
        index_thread_id,
        SessionSource::Cli,
        vec![user_message("index question")],
    );
    let guardian_thread_id = ThreadId::new();
    let unnamed_guardian_thread_id = ThreadId::new();
    for thread_id in [guardian_thread_id, unnamed_guardian_thread_id] {
        write_rollout(
            home.path(),
            thread_id,
            SessionSource::SubAgent(SubAgentSource::Other("guardian".to_string())),
            vec![user_message("large synthetic Guardian prompt")],
        );
    }
    let store = indexed_store(home.path()).await;
    store
        .update_thread_metadata(UpdateThreadMetadataParams {
            thread_id: title_thread_id,
            patch: ThreadMetadataPatch {
                name: Some(Some("renamed title".to_string())),
                ..Default::default()
            },
            include_archived: false,
        })
        .await
        .expect("rename legacy thread");
    codex_rollout::append_thread_name(home.path(), title_thread_id, "stale index title")
        .await
        .expect("write stale legacy index name");
    codex_rollout::append_thread_name(home.path(), index_thread_id, "indexed title")
        .await
        .expect("write legacy index name");

    // Older metadata cleanup also seeded the default name on legacy threads.
    let state_db = store.state_db().await.expect("state runtime");
    let mut guardian_metadata = state_db
        .get_thread(guardian_thread_id)
        .await
        .expect("read Guardian metadata")
        .expect("Guardian metadata");
    guardian_metadata.name = Some(codex_state::GUARDIAN_THREAD_TITLE.to_string());
    state_db
        .upsert_thread(&guardian_metadata)
        .await
        .expect("seed legacy Guardian name");
    codex_rollout::append_thread_name(home.path(), guardian_thread_id, "indexed Guardian name")
        .await
        .expect("write legacy Guardian name");

    for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] {
        if history_mode == ThreadHistoryMode::Paginated {
            store
                .migrate_rollouts(apply_options())
                .await
                .expect("migrate named rollouts");
        }
        let page = store
            .list_threads(ListThreadsParams {
                page_size: 10,
                cursor: None,
                sort_key: ThreadSortKey::CreatedAt,
                sort_direction: SortDirection::Desc,
                allowed_sources: Vec::new(),
                model_providers: None,
                cwd_filters: None,
                section: None,
                project_id: None,
                archived: false,
                search_term: None,
                relation_filter: None,
                use_state_db_only: true,
            })
            .await
            .expect("list threads");
        for (thread_id, name) in [
            (title_thread_id, "renamed title"),
            (index_thread_id, "indexed title"),
            (guardian_thread_id, "indexed Guardian name"),
            (unnamed_guardian_thread_id, "Guardian review"),
        ] {
            let listed = page
                .items
                .iter()
                .find(|thread| thread.thread_id == thread_id)
                .expect("listed thread");
            let read = store
                .read_thread(crate::ReadThreadParams {
                    thread_id,
                    include_archived: false,
                    include_history: false,
                })
                .await
                .expect("read thread");
            assert_eq!(
                (
                    listed.name.as_deref(),
                    read.name.as_deref(),
                    listed.history_mode,
                    read.history_mode
                ),
                (Some(name), Some(name), history_mode, history_mode),
            );
        }
    }
}

#[tokio::test]
async fn migration_repairs_a_missing_paginated_name_when_rerun() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("question")],
    );
    let store = indexed_store(home.path()).await;
    store
        .update_thread_metadata(UpdateThreadMetadataParams {
            thread_id,
            patch: ThreadMetadataPatch {
                name: Some(Some("renamed title".to_string())),
                ..Default::default()
            },
            include_archived: false,
        })
        .await
        .expect("rename legacy thread");
    store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate named rollout");
    let state_db = store.state_db().await.expect("state runtime");
    state_db
        .update_thread_title(thread_id, "question")
        .await
        .expect("restore derived title");
    state_db
        .update_thread_name(thread_id, /*name*/ None)
        .await
        .expect("clear migrated name");

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("repair migrated name");

    assert_eq!(
        report.outcomes[0].status,
        RolloutMigrationStatus::AlreadyPaginated
    );
    assert_eq!(
        state_db
            .get_thread(thread_id)
            .await
            .expect("read repaired metadata")
            .expect("repaired thread")
            .name
            .as_deref(),
        Some("renamed title")
    );
}

#[test]
fn cancelled_compression_workers_hold_migration_locks() {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .max_blocking_threads(1)
        .build()
        .expect("build runtime");
    runtime.block_on(async {
        for compress in [true, false] {
            let home = TempDir::new().expect("create Codex home");
            let thread_id = ThreadId::new();
            let plain_path = home.path().join("rollout.jsonl");
            let compressed_path = home.path().join("rollout.jsonl.zst");
            let contents = b"rollout contents\n";
            fs::write(&plain_path, contents).expect("write plain rollout");
            if !compress {
                compress_rollout(&plain_path);
            }
            let source_path = if compress {
                &plain_path
            } else {
                &compressed_path
            };
            let permissions = fs::metadata(source_path)
                .expect("source metadata")
                .permissions();
            let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
            let locks = Arc::new(MigrationLocks {
                _maintenance: Some(Arc::new(
                    codex_rollout::try_acquire_rollout_maintenance_lock(home.path())
                        .expect("acquire maintenance lock")
                        .expect("maintenance lock is available"),
                )),
                _writer: store
                    .acquire_writer_lock(thread_id)
                    .expect("acquire writer lock"),
                _live_writer: store.live_writer_locks.lock(thread_id).await,
            });
            let coordination = store.live_writer_locks.coordination(thread_id).await;

            // Hold the only blocking thread so the real file worker stays queued across cancellation.
            let (reached_tx, reached_rx) = tokio::sync::oneshot::channel();
            let (resume_tx, resume_rx) = std::sync::mpsc::channel::<()>();
            let blocker = tokio::task::spawn_blocking(move || {
                let _ = reached_tx.send(());
                // Dropping the sender after a failed assertion also lets runtime teardown finish.
                let _ = resume_rx.recv();
            });
            reached_rx.await.expect("blocking thread occupied");
            let mut work = Box::pin(async {
                if compress {
                    compress_rollout_to_path(
                        &plain_path,
                        &compressed_path,
                        permissions,
                        /*modified_at*/ None,
                        &locks,
                    )
                    .await
                } else {
                    decompress_rollout_to_path(&compressed_path, &plain_path, &locks).await
                }
            });
            assert!(futures::poll!(&mut work).is_pending());
            drop(work);
            drop(locks);

            assert!(
                codex_rollout::try_acquire_rollout_maintenance_lock(home.path())
                    .expect("probe maintenance lock")
                    .is_none()
            );
            assert!(matches!(
                store.acquire_writer_lock(thread_id),
                Err(ThreadStoreError::Conflict { .. })
            ));
            assert!(coordination.writer.try_lock().is_err());

            drop(resume_tx);
            blocker.await.expect("release blocking thread");
            // This guard is released last, after the worker's file I/O and other locks.
            let _live_writer_guard = coordination.writer.lock().await;
            assert!(
                codex_rollout::try_acquire_rollout_maintenance_lock(home.path())
                    .expect("probe released maintenance lock")
                    .is_some()
            );
            store
                .acquire_writer_lock(thread_id)
                .expect("writer lock released");
            let decoded = if compress {
                zstd::stream::decode_all(
                    fs::File::open(&compressed_path).expect("open compressed rollout"),
                )
                .expect("decode complete rollout")
            } else {
                #[cfg(unix)]
                assert_eq!(
                    fs::metadata(&plain_path)
                        .expect("temporary metadata")
                        .permissions()
                        .mode()
                        & 0o777,
                    0o600
                );
                fs::read(&plain_path).expect("read decompressed rollout")
            };
            assert_eq!(decoded, contents);
        }
    });
}

#[tokio::test]
async fn migration_skips_threads_with_an_active_writer() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("active question")],
    );
    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
    let _writer = store
        .acquire_writer_lock(thread_id)
        .expect("acquire live writer lock");
    let original = fs::read(&path).expect("read active rollout");

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("inspect active writer");

    assert_eq!(
        report.outcomes[0].status,
        RolloutMigrationStatus::SkippedBusy
    );
    assert_eq!(fs::read(&path).expect("read unmodified rollout"), original);
}

#[tokio::test]
async fn migration_apply_conflicts_with_rollout_maintenance() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("maintenance question")],
    );
    let original = fs::read(&path).expect("read legacy rollout");
    let _maintenance_guard = codex_rollout::try_acquire_rollout_maintenance_lock(home.path())
        .expect("acquire rollout maintenance lock")
        .expect("claim rollout maintenance lock");
    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);

    let error = store
        .migrate_rollouts(apply_options())
        .await
        .expect_err("reject concurrent rollout maintenance");

    assert!(matches!(error, ThreadStoreError::Conflict { .. }));
    assert_eq!(fs::read(&path).expect("read untouched rollout"), original);
}

#[tokio::test]
async fn migration_recovers_a_published_rollout_with_missing_projection() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            user_message("recover question"),
            agent_message("recover answer"),
        ],
    );
    let store = indexed_store(home.path()).await;
    store
        .migrate_rollouts(apply_options())
        .await
        .expect("publish canonical rollout");
    thread_history::delete_thread(&store, thread_id)
        .await
        .expect("simulate interrupted projection");
    let journal_path = migration_journal_path(home.path(), thread_id);
    write_migration_journal(&journal_path)
        .await
        .expect("simulate pending migration journal");

    let writer = store
        .acquire_writer_lock(thread_id)
        .expect("acquire live writer lock");
    let busy = store
        .migrate_rollouts(apply_options())
        .await
        .expect("inspect busy published recovery");
    assert_eq!(busy.outcomes[0].status, RolloutMigrationStatus::SkippedBusy);
    assert!(journal_path.exists());
    drop(writer);

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("recover published rollout");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(!journal_path.exists());
    let projection = thread_history::projection_state(&store, thread_id)
        .await
        .expect("read repaired projection")
        .expect("projection was rebuilt");
    assert_eq!(
        projection.next_byte_offset,
        fs::metadata(&path).expect("read rollout metadata").len()
    );
}

#[tokio::test]
async fn migration_recovers_pending_rollouts_before_new_work() {
    let home = TempDir::new().expect("create Codex home");
    let pending_thread_id = ThreadId::new();
    write_rollout(
        home.path(),
        pending_thread_id,
        SessionSource::Cli,
        vec![user_message("pending question")],
    );
    let new_thread_id = ThreadId::new();
    let new_path = write_rollout(
        home.path(),
        new_thread_id,
        SessionSource::Cli,
        vec![user_message("new question")],
    );
    let newer_directory = home.path().join("sessions/2025/01/04");
    fs::create_dir_all(&newer_directory).expect("create newer rollout directory");
    fs::rename(
        &new_path,
        newer_directory.join(new_path.file_name().expect("rollout filename")),
    )
    .expect("move newer rollout");
    let store = indexed_store(home.path()).await;

    store
        .migrate_rollouts(RolloutMigrationOptions {
            thread_ids: vec![pending_thread_id],
            ..apply_options()
        })
        .await
        .expect("publish pending rollout");
    thread_history::delete_thread(&store, pending_thread_id)
        .await
        .expect("simulate missing projection");
    write_migration_journal(&migration_journal_path(home.path(), pending_thread_id))
        .await
        .expect("simulate pending migration journal");

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("recover pending rollout before new work");

    assert_eq!(
        report
            .outcomes
            .iter()
            .map(|outcome| (
                outcome.thread_id.expect("rollout thread ID"),
                outcome.status
            ))
            .collect::<Vec<_>>(),
        vec![
            (pending_thread_id, RolloutMigrationStatus::Migrated),
            (new_thread_id, RolloutMigrationStatus::Migrated),
        ]
    );
}

#[tokio::test]
async fn migration_recovers_a_compressed_published_rollout() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![
            user_message("recover compressed question"),
            agent_message("recover compressed answer"),
        ],
    );
    let store = indexed_store(home.path()).await;
    store
        .migrate_rollouts(apply_options())
        .await
        .expect("publish canonical rollout");
    thread_history::delete_thread(&store, thread_id)
        .await
        .expect("simulate interrupted projection");
    let state_db = store.state_db.as_ref().expect("state db");
    let mut legacy_metadata = state_db
        .get_thread(thread_id)
        .await
        .expect("read thread metadata")
        .expect("thread metadata");
    legacy_metadata.history_mode = ThreadHistoryMode::Legacy;
    state_db
        .delete_thread(thread_id)
        .await
        .expect("remove paginated thread metadata");
    assert!(
        state_db
            .insert_thread_if_absent(&legacy_metadata)
            .await
            .expect("restore legacy thread metadata")
    );
    let journal_path = migration_journal_path(home.path(), thread_id);
    write_migration_journal(&journal_path)
        .await
        .expect("simulate pending migration journal");
    let compressed_path = compress_rollout(&path);

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("recover compressed published rollout");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(!path.exists());
    assert!(compressed_path.exists());
    assert!(!journal_path.exists());
    assert_eq!(
        store
            .state_db
            .as_ref()
            .expect("state db")
            .get_thread(thread_id)
            .await
            .expect("read repaired thread metadata")
            .expect("thread metadata")
            .history_mode,
        ThreadHistoryMode::Paginated
    );
    let projection = thread_history::projection_state(&store, thread_id)
        .await
        .expect("read repaired projection")
        .expect("projection was rebuilt");
    assert_eq!(
        projection.next_byte_offset,
        zstd::stream::decode_all(fs::File::open(&compressed_path).expect("open rollout"))
            .expect("decompress rollout")
            .len() as u64
    );
}

#[tokio::test]
async fn migration_skips_oversized_jsonl_records() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("kept question")],
    );
    let oversized_output = "x".repeat(super::MAX_ROLLOUT_LINE_BYTES);
    let mut file = fs::OpenOptions::new()
        .append(true)
        .open(&path)
        .expect("open legacy rollout");
    writeln!(
        file,
        "{{\"timestamp\":\"{TIMESTAMP}\",\"type\":\"response_item\",\"payload\":{{\"type\":\"function_call_output\",\"call_id\":\"call-1\",\"output\":\"{oversized_output}\"}}}}"
    )
    .expect("write oversized rollout record");
    drop(file);
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate oversized rollout record");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(
        fs::metadata(&path)
            .expect("read migrated rollout metadata")
            .len()
            < super::MAX_ROLLOUT_LINE_BYTES as u64
    );
    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(turns.turns.len(), 1);
    assert_eq!(turns.turns[0].items.len(), 1);
}

#[tokio::test]
async fn migration_skips_empty_rollout_files() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let directory = home.path().join("sessions/2025/01/03");
    fs::create_dir_all(&directory).expect("create rollout directory");
    let path = directory.join(format!("rollout-2025-01-03T12-00-00-{thread_id}.jsonl"));
    fs::File::create(&path).expect("create empty rollout");
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("inspect empty rollout");

    assert_eq!(
        report.outcomes[0].status,
        RolloutMigrationStatus::SkippedEmpty
    );
    assert_eq!(fs::metadata(&path).expect("read empty rollout").len(), 0);
    assert!(
        store
            .state_db
            .as_ref()
            .expect("state db")
            .get_thread(thread_id)
            .await
            .expect("read thread metadata")
            .is_none()
    );
}

#[tokio::test]
async fn selected_migration_prunes_unrelated_indexed_rollouts() {
    let home = TempDir::new().expect("create Codex home");
    let owner = ThreadId::new();
    let original = write_rollout(home.path(), owner, SessionSource::Cli, vec![]);
    let directory = original.parent().expect("rollout directory");
    let legacy = directory.join("rollout-imported.jsonl");
    let reverted = directory.join(format!(
        "rollout-2025-01-03T12-00-00-{owner}_{}.jsonl",
        ThreadId::new()
    ));
    fs::copy(&original, &legacy).expect("copy legacy rollout");
    fs::rename(&original, &reverted).expect("rename reverted rollout");
    let reverted = compress_rollout(&reverted);
    let other = write_rollout(home.path(), ThreadId::new(), SessionSource::Cli, vec![]);
    fs::copy(&other, &original).expect("put another owner's metadata under the selected filename");
    let archived_owner = ThreadId::new();
    let archived_reverted = directory.join(format!(
        "rollout-2025-01-03T12-00-00-{archived_owner}_{}.jsonl",
        ThreadId::new()
    ));
    let archived = write_rollout(home.path(), archived_owner, SessionSource::Cli, vec![]);
    fs::rename(archived, &archived_reverted).expect("rename unrelated reverted rollout");
    let archived_reverted = move_to_archived(home.path(), archived_reverted);
    let misnamed = directory.join(format!(
        "rollout-2025-01-03T12-00-00-{}.jsonl",
        ThreadId::new()
    ));
    fs::copy(&legacy, &misnamed).expect("copy under a different canonical thread ID");
    let config = test_config(home.path());
    let db = codex_state::StateRuntime::init(config.sqlite.clone(), "test-provider".to_string())
        .await
        .expect("initialize SQLite");
    for path in [&other, &archived_reverted, &reverted, &misnamed] {
        let mut row = codex_rollout::builder_from_items(&[], path)
            .expect("read indexed filename")
            .build("test-provider");
        if path == &misnamed {
            row.id = ThreadId::new();
        }
        db.upsert_thread(&row).await.expect("index rollout");
    }
    db.close().await;
    // SQLite still points to the plain file after compression.
    compress_rollout(&archived_reverted);
    let store = LocalThreadStore::new(config, /*state_db*/ None);
    let mut inspected = 0;
    let report = store
        .migrate_rollouts_with_progress(
            RolloutMigrationOptions {
                thread_ids: vec![owner],
                ..Default::default()
            },
            |_| inspected += 1,
        )
        .await
        .expect("inspect selected rollouts");
    let mut actual = report
        .outcomes
        .into_iter()
        .map(|outcome| outcome.rollout_path)
        .collect::<Vec<_>>();
    let mut expected = vec![legacy, reverted, misnamed];
    actual.sort();
    expected.sort();
    assert_eq!(actual, expected);
    // The unindexed file with another owner's header also needs inspection.
    assert_eq!(inspected, expected.len() + 1);
}

#[tokio::test]
async fn migration_reports_missing_sqlite_metadata() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("question")],
    );
    let store = indexed_store(home.path()).await;
    store
        .state_db
        .as_ref()
        .expect("state db")
        .delete_thread(thread_id)
        .await
        .expect("remove thread metadata");

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("inspect rollout with missing metadata");

    assert_failed_with_reason(
        &report.outcomes[0],
        RolloutMigrationFailureReason::MissingSqliteMetadata,
    );
}

#[tokio::test]
async fn migration_reports_invalid_session_metadata() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let directory = home.path().join("sessions/2025/01/03");
    fs::create_dir_all(&directory).expect("create rollout directory");
    let path = directory.join(format!("rollout-2025-01-03T12-00-00-{thread_id}.jsonl"));
    fs::write(path, "not a rollout record\n").expect("write malformed rollout");
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("inspect rollout with invalid metadata");

    assert_failed_with_reason(
        &report.outcomes[0],
        RolloutMigrationFailureReason::InvalidSessionMetadata,
    );
}

#[tokio::test]
async fn migration_skips_malformed_lines_and_trailing_partial_tail() {
    let home = TempDir::new().expect("create Codex home");
    let thread_id = ThreadId::new();
    let path = write_rollout(
        home.path(),
        thread_id,
        SessionSource::Cli,
        vec![user_message("kept question")],
    );
    let mut file = fs::OpenOptions::new()
        .append(true)
        .open(&path)
        .expect("open legacy rollout");
    writeln!(file, "{{not valid rollout json").expect("append malformed record");
    let later_valid_line = RolloutLine {
        timestamp: TIMESTAMP.to_string(),
        ordinal: None,
        item: agent_message("kept answer"),
    };
    writeln!(
        file,
        "{}",
        serde_json::to_string(&later_valid_line).expect("serialize later valid record")
    )
    .expect("append later valid record");
    file.write_all(br#"{"timestamp":"unterminated""#)
        .expect("append partial tail");
    drop(file);
    let store = indexed_store(home.path()).await;

    let report = store
        .migrate_rollouts(apply_options())
        .await
        .expect("migrate malformed legacy rollout");

    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
    assert!(!migration_journal_path(home.path(), thread_id).exists());
    let turns = list_active_summary_turns(&store, thread_id).await;
    assert_eq!(turns.turns.len(), 1);
    assert_eq!(turns.turns[0].items.len(), 2);
}