From 8b57fdd973a501e8a1fa4fa1b2cd61952dc65157 Mon Sep 17 00:00:00 2001 From: Yaroslav Volovich Date: Mon, 23 Feb 2026 13:47:41 +0000 Subject: [PATCH] app-server: support turn-based thread forks --- .../schema/typescript/v2/ThreadForkParams.ts | 4 + .../src/protocol/thread_history.rs | 100 ++++- .../app-server-protocol/src/protocol/v2.rs | 6 + codex-rs/app-server/README.md | 12 +- .../app-server/src/codex_message_processor.rs | 109 ++++- .../app-server/tests/suite/v2/thread_fork.rs | 398 ++++++++++++++++++ 6 files changed, 618 insertions(+), 11 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts index 742e4d7032..31cfb977f1 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts @@ -19,6 +19,10 @@ export type ThreadForkParams = {threadId: string, /** * If specified, the thread_id param will be ignored. */ path?: string | null, /** + * [UNSTABLE] Fork after the specified historical turn (inclusive). + * When omitted, the full thread history is copied. + */ +forkAfterTurnId?: string | null, /** * Configuration overrides for the forked thread, if any. */ model?: string | null, modelProvider?: string | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, /** diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 97ad8d521f..83378c4f69 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -54,22 +54,40 @@ use codex_protocol::protocol::ExecCommandStatus as CoreExecCommandStatus; #[cfg(test)] use codex_protocol::protocol::PatchApplyStatus as CorePatchApplyStatus; +#[derive(Debug, Clone, PartialEq)] +pub struct ThreadHistoryBuildResult { + pub turns: Vec, + /// True when any turn id had to be synthesized during replay because the + /// rollout history lacked explicit turn lifecycle events. + pub has_synthetic_turn_ids: bool, +} + +/// Convert persisted [`RolloutItem`] entries into a sequence of [`Turn`] values and +/// annotate whether any turn ids were synthesized during replay. +/// +/// When available, this uses `TurnContext.turn_id` as the canonical turn id so +/// resumed/rebuilt thread history preserves the original turn identifiers. +pub fn build_thread_history_from_rollout_items(items: &[RolloutItem]) -> ThreadHistoryBuildResult { + let mut builder = ThreadHistoryBuilder::new(); + for item in items { + builder.handle_rollout_item(item); + } + builder.finish_result() +} + /// Convert persisted [`RolloutItem`] entries into a sequence of [`Turn`] values. /// /// When available, this uses `TurnContext.turn_id` as the canonical turn id so /// resumed/rebuilt thread history preserves the original turn identifiers. pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { - let mut builder = ThreadHistoryBuilder::new(); - for item in items { - builder.handle_rollout_item(item); - } - builder.finish() + build_thread_history_from_rollout_items(items).turns } pub struct ThreadHistoryBuilder { turns: Vec, current_turn: Option, next_item_index: i64, + has_synthetic_turn_ids: bool, } impl Default for ThreadHistoryBuilder { @@ -84,6 +102,7 @@ impl ThreadHistoryBuilder { turns: Vec::new(), current_turn: None, next_item_index: 1, + has_synthetic_turn_ids: false, } } @@ -91,9 +110,16 @@ impl ThreadHistoryBuilder { *self = Self::new(); } - pub fn finish(mut self) -> Vec { + pub fn finish(self) -> Vec { + self.finish_result().turns + } + + pub fn finish_result(mut self) -> ThreadHistoryBuildResult { self.finish_current_turn(); - self.turns + ThreadHistoryBuildResult { + turns: self.turns, + has_synthetic_turn_ids: self.has_synthetic_turn_ids, + } } pub fn active_turn_snapshot(&self) -> Option { @@ -861,8 +887,15 @@ impl ThreadHistoryBuilder { } fn new_turn(&mut self, id: Option) -> PendingTurn { + let id = match id { + Some(id) => id, + None => { + self.has_synthetic_turn_ids = true; + Uuid::now_v7().to_string() + } + }; PendingTurn { - id: id.unwrap_or_else(|| Uuid::now_v7().to_string()), + id, items: Vec::new(), error: None, status: TurnStatus::Completed, @@ -1198,6 +1231,57 @@ mod tests { ); } + #[test] + fn reports_synthetic_turn_ids_for_legacy_history() { + let items = vec![ + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "legacy".into(), + images: None, + text_elements: Vec::new(), + local_images: Vec::new(), + })), + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: "reply".into(), + phase: None, + })), + ]; + + let result = build_thread_history_from_rollout_items(&items); + assert_eq!(result.turns.len(), 1); + assert_eq!(result.has_synthetic_turn_ids, true); + } + + #[test] + fn reports_no_synthetic_turn_ids_when_turn_boundaries_are_explicit() { + let turn_id = "turn-explicit"; + let items = vec![ + RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_id: turn_id.to_string(), + model_context_window: Some(128_000), + collaboration_mode_kind: Default::default(), + })), + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "modern".into(), + images: None, + text_elements: Vec::new(), + local_images: Vec::new(), + })), + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: "reply".into(), + phase: None, + })), + RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: turn_id.to_string(), + last_agent_message: Some("reply".into()), + })), + ]; + + let result = build_thread_history_from_rollout_items(&items); + assert_eq!(result.turns.len(), 1); + assert_eq!(result.turns[0].id, turn_id); + assert_eq!(result.has_synthetic_turn_ids, false); + } + #[test] fn ignores_non_plan_item_lifecycle_events() { let turn_id = "turn-1"; diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 862460a4a7..00ca885bf4 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1869,6 +1869,12 @@ pub struct ThreadForkParams { #[ts(optional = nullable)] pub path: Option, + /// [UNSTABLE] Fork after the specified historical turn (inclusive). + /// When omitted, the full thread history is copied. + #[experimental("thread/fork.forkAfterTurnId")] + #[ts(optional = nullable)] + pub fork_after_turn_id: Option, + /// Configuration overrides for the forked thread, if any. #[ts(optional = nullable)] pub model: Option, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 8a77f42339..44b479ce31 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -121,7 +121,7 @@ Example with notification opt-out: - `thread/start` — create a new thread; emits `thread/started` and auto-subscribes you to turn/item events for that thread. - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. -- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; emits `thread/started` and auto-subscribes you to turn/item events for the new thread. +- `thread/fork` — fork an existing thread into a new thread id by copying the stored history. Experimental `forkAfterTurnId` lets clients fork from a selected historical turn (modern turn-id-stable histories only); emits `thread/started` and auto-subscribes you to turn/item events for the new thread. - `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. - `thread/loaded/list` — list the thread ids currently loaded in memory. - `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. @@ -215,7 +215,7 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev { "id": 11, "result": { "thread": { "id": "thr_123", … } } } ``` -To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it: +To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it. To fork from a selected prior turn instead of the full history, pass experimental `forkAfterTurnId`: ```json { "method": "thread/fork", "id": 12, "params": { "threadId": "thr_123" } } @@ -223,6 +223,14 @@ To branch from a stored session, call `thread/fork` with the `thread.id`. This c { "method": "thread/started", "params": { "thread": { … } } } ``` +```json +{ "method": "thread/fork", "id": 13, "params": { + "threadId": "thr_123", + "forkAfterTurnId": "turn_abc", + "cwd": "/Users/me/project-worktree" +} } +``` + Experimental API: `thread/start`, `thread/resume`, and `thread/fork` accept `persistExtendedHistory: true` to persist a richer subset of ThreadItems for non-lossy history when calling `thread/read`, `thread/resume`, and `thread/fork` later. This does not backfill events that were not persisted previously. ### Example: List threads (with pagination & filters) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 1897898718..88d9a885a9 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -175,6 +175,7 @@ use codex_app_server_protocol::WindowsSandboxSetupCompletedNotification; use codex_app_server_protocol::WindowsSandboxSetupMode; use codex_app_server_protocol::WindowsSandboxSetupStartParams; use codex_app_server_protocol::WindowsSandboxSetupStartResponse; +use codex_app_server_protocol::build_thread_history_from_rollout_items; use codex_app_server_protocol::build_turns_from_rollout_items; use codex_arg0::Arg0DispatchPaths; use codex_backend_client::Client as BackendClient; @@ -3324,6 +3325,7 @@ impl CodexMessageProcessor { let ThreadForkParams { thread_id, path, + fork_after_turn_id, model, model_provider, cwd, @@ -3437,6 +3439,48 @@ impl CodexMessageProcessor { }; let fallback_model_provider = config.model_provider_id.clone(); + let fork_cutoff_nth_user_message = if let Some(fork_after_turn_id) = + fork_after_turn_id.as_deref() + { + let source_items = match read_rollout_items_from_rollout(rollout_path.as_path()).await { + Ok(items) => items, + Err(err) => { + let (code, message) = match err.kind() { + std::io::ErrorKind::NotFound => ( + INVALID_REQUEST_ERROR_CODE, + format!("failed to load rollout `{}`: {err}", rollout_path.display()), + ), + _ => ( + INTERNAL_ERROR_CODE, + format!( + "failed to read rollout `{}` for thread fork: {err}", + rollout_path.display() + ), + ), + }; + let error = JSONRPCErrorError { + code, + message, + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + match resolve_thread_fork_cutoff_nth_user_message( + source_items.as_slice(), + fork_after_turn_id, + ) { + Ok(cutoff) => cutoff, + Err(message) => { + self.send_invalid_request_error(request_id, message).await; + return; + } + } + } else { + usize::MAX + }; let NewThread { thread_id, @@ -3445,7 +3489,7 @@ impl CodexMessageProcessor { } = match self .thread_manager .fork_thread( - usize::MAX, + fork_cutoff_nth_user_message, config, rollout_path.clone(), persist_extended_history, @@ -6869,6 +6913,69 @@ async fn sync_default_client_residency_requirement( } } +fn resolve_thread_fork_cutoff_nth_user_message( + rollout_items: &[RolloutItem], + fork_after_turn_id: &str, +) -> Result { + let history = build_thread_history_from_rollout_items(rollout_items); + if history.has_synthetic_turn_ids { + return Err( + "turn-based forking is not supported for legacy thread history; use full thread/fork" + .to_string(), + ); + } + + let total_user_turns = history + .turns + .iter() + .filter(|turn| { + turn.items + .iter() + .any(|item| matches!(item, ThreadItem::UserMessage { .. })) + }) + .count(); + let mut user_turn_index = 0usize; + + for turn in &history.turns { + let has_user_message = turn + .items + .iter() + .any(|item| matches!(item, ThreadItem::UserMessage { .. })); + let has_agent_message = turn + .items + .iter() + .any(|item| matches!(item, ThreadItem::AgentMessage { .. })); + + if turn.id == fork_after_turn_id { + if matches!(turn.status, TurnStatus::InProgress) { + return Err( + "fork turn must be completed/interrupted/failed, not in progress".to_string(), + ); + } + if !has_user_message { + return Err("fork turn must contain a user message".to_string()); + } + if !has_agent_message { + return Err("fork turn must contain an agent message".to_string()); + } + + return if user_turn_index.saturating_add(1) >= total_user_turns { + Ok(usize::MAX) + } else { + Ok(user_turn_index.saturating_add(1)) + }; + } + + if has_user_message { + user_turn_index = user_turn_index.saturating_add(1); + } + } + + Err(format!( + "fork turn not found in source thread history: {fork_after_turn_id}" + )) +} + /// Derive the effective [`Config`] by layering three override sources. /// /// Precedence (lowest to highest): diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index a6f9942171..f96908073c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -2,6 +2,7 @@ use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_fake_rollout; use app_test_support::create_mock_responses_server_repeating_assistant; +use app_test_support::rollout_path; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCNotification; @@ -17,11 +18,26 @@ use codex_app_server_protocol::ThreadStartedNotification; use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; +use codex_protocol::ThreadId; +use codex_protocol::models::MessagePhase; +use codex_protocol::protocol::AgentMessageEvent; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnStartedEvent; +use codex_protocol::protocol::UserMessageEvent; use pretty_assertions::assert_eq; use serde_json::Value; +use serde_json::json; +use std::fs; +use std::fs::FileTimes; use std::path::Path; +use std::path::PathBuf; use tempfile::TempDir; use tokio::time::timeout; +use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); @@ -191,6 +207,234 @@ async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_fork_can_fork_after_selected_turn() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let conversation_id = create_fake_rollout_with_explicit_turns( + codex_home.path(), + "2025-01-05T12-00-01", + "2025-01-05T12:00:01Z", + &[ + ExplicitTurnFixture { + turn_id: "turn-1", + user_text: "u1", + agent_text: Some("a1"), + state: FixtureTurnState::Completed, + }, + ExplicitTurnFixture { + turn_id: "turn-2", + user_text: "u2", + agent_text: Some("a2"), + state: FixtureTurnState::Completed, + }, + ExplicitTurnFixture { + turn_id: "turn-3", + user_text: "u3", + agent_text: Some("a3"), + state: FixtureTurnState::Completed, + }, + ], + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_cwd = codex_home.path().join("fork-worktree"); + fs::create_dir_all(&fork_cwd)?; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id, + fork_after_turn_id: Some("turn-2".to_string()), + cwd: Some(fork_cwd.display().to_string()), + ..Default::default() + }) + .await?; + let fork_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(fork_id)), + ) + .await??; + let ThreadForkResponse { thread, cwd, .. } = to_response::(fork_resp)?; + + assert_eq!(cwd, fork_cwd); + assert_eq!(thread.cwd, fork_cwd); + assert_eq!(thread.turns.len(), 2, "later turns should be truncated"); + assert_eq!(thread.turns[0].id, "turn-1"); + assert_eq!(thread.turns[1].id, "turn-2"); + assert_turn_user_text(&thread.turns[0].items, "u1"); + assert_turn_user_text(&thread.turns[1].items, "u2"); + Ok(()) +} + +#[tokio::test] +async fn thread_fork_rejects_unknown_turn_anchor() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let conversation_id = create_fake_rollout_with_explicit_turns( + codex_home.path(), + "2025-01-05T12-00-02", + "2025-01-05T12:00:02Z", + &[ExplicitTurnFixture { + turn_id: "turn-1", + user_text: "u1", + agent_text: Some("a1"), + state: FixtureTurnState::Completed, + }], + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id, + fork_after_turn_id: Some("missing-turn".to_string()), + ..Default::default() + }) + .await?; + let fork_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + ) + .await??; + + assert!( + fork_err.error.message.contains("fork turn not found"), + "unexpected fork error: {}", + fork_err.error.message + ); + Ok(()) +} + +#[tokio::test] +async fn thread_fork_rejects_legacy_turn_anchor() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-03", + "2025-01-05T12:00:03Z", + "legacy preview", + Some("mock_provider"), + None, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id, + fork_after_turn_id: Some("legacy-turn".to_string()), + ..Default::default() + }) + .await?; + let fork_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + ) + .await??; + + assert!( + fork_err.error.message.contains("legacy thread history"), + "unexpected fork error: {}", + fork_err.error.message + ); + Ok(()) +} + +#[tokio::test] +async fn thread_fork_rejects_in_progress_turn_anchor() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let conversation_id = create_fake_rollout_with_explicit_turns( + codex_home.path(), + "2025-01-05T12-00-04", + "2025-01-05T12:00:04Z", + &[ExplicitTurnFixture { + turn_id: "turn-in-progress", + user_text: "u1", + agent_text: Some("a1"), + state: FixtureTurnState::InProgress, + }], + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id, + fork_after_turn_id: Some("turn-in-progress".to_string()), + ..Default::default() + }) + .await?; + let fork_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + ) + .await??; + + assert!( + fork_err.error.message.contains("not in progress"), + "unexpected fork error: {}", + fork_err.error.message + ); + Ok(()) +} + +#[tokio::test] +async fn thread_fork_rejects_turn_anchor_without_agent_message() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let conversation_id = create_fake_rollout_with_explicit_turns( + codex_home.path(), + "2025-01-05T12-00-05", + "2025-01-05T12:00:05Z", + &[ExplicitTurnFixture { + turn_id: "turn-no-agent", + user_text: "u1", + agent_text: None, + state: FixtureTurnState::Completed, + }], + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id, + fork_after_turn_id: Some("turn-no-agent".to_string()), + ..Default::default() + }) + .await?; + let fork_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(fork_id)), + ) + .await??; + + assert!( + fork_err.error.message.contains("agent message"), + "unexpected fork error: {}", + fork_err.error.message + ); + Ok(()) +} + // Helper to create a config.toml pointing at the mock model server. fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); @@ -214,3 +458,157 @@ stream_max_retries = 0 ), ) } + +#[derive(Clone, Copy)] +enum FixtureTurnState { + Completed, + InProgress, +} + +#[derive(Clone, Copy)] +struct ExplicitTurnFixture<'a> { + turn_id: &'a str, + user_text: &'a str, + agent_text: Option<&'a str>, + state: FixtureTurnState, +} + +fn assert_turn_user_text(items: &[ThreadItem], expected: &str) { + match items.first() { + Some(ThreadItem::UserMessage { content, .. }) => assert_eq!( + content, + &vec![UserInput::Text { + text: expected.to_string(), + text_elements: Vec::new(), + }] + ), + other => panic!("expected first turn item to be a user message, got {other:?}"), + } +} + +fn create_fake_rollout_with_explicit_turns( + codex_home: &Path, + filename_ts: &str, + meta_rfc3339: &str, + turns: &[ExplicitTurnFixture<'_>], +) -> Result { + let uuid = Uuid::new_v4(); + let uuid_str = uuid.to_string(); + let conversation_id = ThreadId::from_string(&uuid_str)?; + let file_path = rollout_path(codex_home, filename_ts, &uuid_str); + let dir = file_path + .parent() + .ok_or_else(|| anyhow::anyhow!("missing rollout parent directory"))?; + fs::create_dir_all(dir)?; + + let meta = SessionMeta { + id: conversation_id, + forked_from_id: None, + timestamp: meta_rfc3339.to_string(), + cwd: PathBuf::from("/"), + originator: "codex".to_string(), + cli_version: "0.0.0".to_string(), + source: codex_protocol::protocol::SessionSource::Cli, + agent_nickname: None, + agent_role: None, + model_provider: Some("mock_provider".to_string()), + base_instructions: None, + dynamic_tools: None, + }; + let mut lines = vec![rollout_line( + meta_rfc3339, + RolloutItem::SessionMeta(SessionMetaLine { meta, git: None }), + )?]; + + for (idx, turn) in turns.iter().enumerate() { + lines.push( + json!({ + "timestamp": meta_rfc3339, + "type":"response_item", + "payload": { + "type":"message", + "role":"user", + "content":[{"type":"input_text","text": turn.user_text}] + } + }) + .to_string(), + ); + + lines.push(rollout_line( + meta_rfc3339, + RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent { + turn_id: turn.turn_id.to_string(), + model_context_window: Some(128_000), + collaboration_mode_kind: Default::default(), + })), + )?); + lines.push(rollout_line( + meta_rfc3339, + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: turn.user_text.to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })), + )?); + if let Some(agent_text) = turn.agent_text { + lines.push(rollout_line( + meta_rfc3339, + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: agent_text.to_string(), + phase: Some(MessagePhase::FinalAnswer), + })), + )?); + } + if matches!(turn.state, FixtureTurnState::Completed) { + let last_agent_message = turn.agent_text.map(str::to_string); + lines.push(rollout_line( + meta_rfc3339, + RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: turn.turn_id.to_string(), + last_agent_message, + })), + )?); + } + + if idx == 0 { + lines.push( + json!({ + "timestamp": meta_rfc3339, + "type":"response_item", + "payload": { + "type":"message", + "role":"assistant", + "content":[{"type":"output_text","text": turn.agent_text.unwrap_or("")}] + } + }) + .to_string(), + ); + } + } + + fs::write(&file_path, lines.join("\n") + "\n")?; + let parsed = chrono::DateTime::parse_from_rfc3339(meta_rfc3339)?.with_timezone(&chrono::Utc); + let times = FileTimes::new().set_modified(parsed.into()); + fs::OpenOptions::new() + .append(true) + .open(&file_path)? + .set_times(times)?; + Ok(uuid_str) +} + +fn rollout_line(timestamp: &str, item: RolloutItem) -> Result { + let mut line = serde_json::Map::new(); + line.insert( + "timestamp".to_string(), + Value::String(timestamp.to_string()), + ); + + let item_value = serde_json::to_value(item)?; + let Value::Object(item_map) = item_value else { + anyhow::bail!("rollout item did not serialize as an object"); + }; + line.extend(item_map); + + Ok(Value::Object(line).to_string()) +}