diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 841bde32dd..483b244c01 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index b31dfdf093..1279bb8661 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -624,7 +624,7 @@ pub struct ThreadForkParams { #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub ephemeral: bool, /// Persist the fork as an attached thread of its resolved source, hidden from ordinary - /// lists. The source cannot be deleted while attached threads exist. Requires local storage; + /// lists. The source cannot be deleted while attached threads exist. Requires a paginated source and local storage; /// incompatible with ephemeral and deferGoalContinuation. Omitted or false creates an ordinary fork. #[experimental("thread/fork.attachToSource")] #[serde(default, skip_serializing_if = "std::ops::Not::not")] diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index f5f0665e4e..64f8bd39e8 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -4514,7 +4514,8 @@ impl ThreadRequestProcessor { }; restored_token_usage_turn_id(response_history.get_rollout_items(), turns) }) - .filter(|turn_id| !turn_id.is_empty()); + .filter(|turn_id| !turn_id.is_empty()) + .filter(|_| !paginated_resume || thread.forked_from_id.is_none()); if redact_resume_payloads { redact_thread_resume_payloads(&mut thread.turns); if let Some(initial_turns_page) = initial_turns_page.as_mut() { @@ -4858,7 +4859,8 @@ impl ThreadRequestProcessor { }); restored_token_usage_turn_id(history, turns) }) - .filter(|turn_id| !turn_id.is_empty()); + .filter(|turn_id| !turn_id.is_empty()) + .filter(|_| !paginated_resume || thread_summary.forked_from_id.is_none()); let paginated_initial_turns_page_with_active_slot = if paginated_resume { match params.initial_turns_page.as_ref() { Some(params) @@ -5391,6 +5393,11 @@ impl ThreadRequestProcessor { None }; let paginated_source = matches!(source_thread.history_mode, ThreadHistoryMode::Paginated); + if attach_to_source && !paginated_source { + return Err(invalid_request( + "attachToSource requires a paginated parent", + )); + } if last_turn_id.is_some() && before_turn_id.is_some() { return Err(invalid_request( "`beforeTurnId` cannot be combined with `lastTurnId`", @@ -5839,6 +5846,8 @@ impl ThreadRequestProcessor { thread.turns.as_slice(), )); } + let token_usage_turn_id = + token_usage_turn_id.filter(|id| thread.turns.iter().any(|turn| turn.id == *id)); if let Some(name) = source_thread_name { set_thread_name_from_title(&mut thread, name); } diff --git a/codex-rs/app-server/src/request_processors/token_usage_replay.rs b/codex-rs/app-server/src/request_processors/token_usage_replay.rs index c6d21f4cfe..212400a826 100644 --- a/codex-rs/app-server/src/request_processors/token_usage_replay.rs +++ b/codex-rs/app-server/src/request_processors/token_usage_replay.rs @@ -39,6 +39,9 @@ pub(super) async fn send_thread_token_usage_update_to_connection( conversation: &CodexThread, token_usage_turn_id: String, ) { + if token_usage_turn_id.is_empty() { + return; + } let Some(info) = conversation.token_usage_info().await else { return; }; diff --git a/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs b/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs index 5f1b053276..08c95a748b 100644 --- a/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/attached_thread_tests.rs @@ -10,10 +10,10 @@ use codex_app_server_protocol::ThreadDeleteParams; use codex_app_server_protocol::ThreadDeleteResponse; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; -use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; @@ -32,24 +32,16 @@ enum Recovery { Corrupt, } -#[test_case(ThreadHistoryMode::Legacy, Recovery::Restart; "legacy restart")] -#[test_case(ThreadHistoryMode::Paginated, Recovery::Restart; "paginated restart")] -#[test_case(ThreadHistoryMode::Legacy, Recovery::Missing; "legacy missing database")] -#[test_case(ThreadHistoryMode::Paginated, Recovery::Corrupt; "paginated corrupt database")] +#[test_case(Recovery::Restart; "paginated restart")] +#[test_case(Recovery::Missing; "paginated missing database")] +#[test_case(Recovery::Corrupt; "paginated corrupt database")] #[tokio::test] -async fn saved_attached_thread_upgrade_restart_and_delete( - mode: ThreadHistoryMode, - recovery: Recovery, -) -> Result<()> { +async fn saved_attached_thread_upgrade_restart_and_delete(recovery: Recovery) -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let home = TempDir::new()?; MockResponsesConfig::new(&server.uri()).write(home.path())?; // Both parents were persisted before this app-server (and its SQLite migration) starts. - let create_rollout = match mode { - ThreadHistoryMode::Legacy => create_fake_rollout, - ThreadHistoryMode::Paginated => create_fake_paginated_rollout, - }; - let parent = create_rollout( + let parent = create_fake_paginated_rollout( home.path(), "2025-01-05T12-00-00", "2025-01-05T12:00:00Z", @@ -57,7 +49,7 @@ async fn saved_attached_thread_upgrade_restart_and_delete( Some("mock_provider"), /*git_info*/ None, )?; - let other = create_rollout( + let other = create_fake_paginated_rollout( home.path(), "2025-01-06T12-00-00", "2025-01-06T12:00:00Z", @@ -179,13 +171,14 @@ async fn saved_attached_thread_upgrade_restart_and_delete( .send_raw_request( "thread/resume", Some(json!({ - "threadId": children[0], "excludeTurns": true, + "threadId": children[0], "developerInstructions": "Follow BIRCH-849." })), ) .await?; let resumed: ThreadResumeResponse = mcp.read_response(id).await?; assert_eq!(resumed.thread.id, children[0]); + assert!(resumed.thread.turns.is_empty()); // Ordinary forks do not inherit attachment. let id = mcp .send_thread_fork_request(ThreadForkParams { @@ -218,7 +211,7 @@ async fn saved_attached_thread_upgrade_restart_and_delete( ..Default::default() }) .await?; - let _: TurnStartResponse = mcp.read_response(id).await?; + let started: TurnStartResponse = mcp.read_response(id).await?; mcp.read_stream_until_notification_message("turn/completed") .await?; let requests = server.received_requests().await.expect("response requests"); @@ -232,6 +225,19 @@ async fn saved_attached_thread_upgrade_restart_and_delete( input.contains("BIRCH-849"), "resume applies client instructions" ); + assert!( + input.contains("BLUE HERON 72"), + "model recalls the parent after restart" + ); + let id = mcp + .send_raw_request( + "thread/read", + Some(json!({"threadId": children[0], "includeTurns": true})), + ) + .await?; + let read: ThreadReadResponse = mcp.read_response(id).await?; + assert_eq!(read.thread.turns.len(), 1); + assert_eq!(read.thread.turns[0].id, started.turn.id); let id = mcp .send_raw_request("thread/resume", Some(json!({"threadId": parent}))) .await?; @@ -296,6 +302,11 @@ async fn saved_attached_thread_rejects_conflicting_fork_and_discovery_filters() .build_initialized() .await?; for (method, params, message) in [ + ( + "thread/fork", + json!({"threadId": parent, "attachToSource": true}), + "attachToSource requires a paginated parent", + ), ( "thread/fork", json!({"threadId": parent, "attachToSource": true, "ephemeral": true}), 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 146cadfda3..355f25ba84 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1860,18 +1860,13 @@ async fn thread_fork_warns_for_paginated_full_history_hydration() -> Result<()> Ok(()) } +#[test_case::test_case(MultiAgentVersion::V1, false; "ordinary v1")] +#[test_case::test_case(MultiAgentVersion::V2, false; "ordinary v2")] +#[test_case::test_case(MultiAgentVersion::V1, true; "attached")] #[tokio::test] -async fn thread_fork_freezes_active_paginated_turn_as_interrupted() -> Result<()> { - assert_thread_fork_freezes_active_paginated_turn_as_interrupted(MultiAgentVersion::V1).await -} - -#[tokio::test] -async fn thread_fork_persists_developer_interruption_marker_for_multi_agent_v2() -> Result<()> { - assert_thread_fork_freezes_active_paginated_turn_as_interrupted(MultiAgentVersion::V2).await -} - -async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( +async fn thread_fork_freezes_active_paginated_turn_as_interrupted( multi_agent_version: MultiAgentVersion, + attach_to_source: bool, ) -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; @@ -1995,6 +1990,7 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( params: ThreadForkParams { thread_id: source_thread_id.clone(), thread_source, + attach_to_source, ..Default::default() }, }) @@ -2108,6 +2104,25 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted( }, }) .await?; + if attach_to_source { + assert!(turns.is_empty()); + for (method, field) in [ + ("thread/revert", "beforeTurnId"), + ("thread/fork", "lastTurnId"), + ] { + let id = mcp + .send_raw_request( + method, + Some(json!({"threadId": forked_thread_id, (field): "active-turn"})), + ) + .await?; + let error = mcp + .read_stream_until_error_message(RequestId::Integer(id)) + .await?; + assert!(error.error.message.contains("turn not found: active-turn")); + } + return Ok(()); + } assert_eq!(turns.len(), 1); assert_eq!(turns[0].id, "active-turn"); assert_eq!(turns[0].root_turn_id.as_deref(), Some("causal-root")); diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 060522cdb7..2b6946e164 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -4302,6 +4302,38 @@ async fn cold_paginated_resume_restores_usage_without_loading_turns() -> Result< timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(turns_id)).await??; assert_eq!(notification.turn_id, turns.data[0].id); + let id = app_server + .send_raw_request( + "thread/fork", + Some(json!({"threadId": notification.thread_id, "attachToSource": true})), + ) + .await?; + let child: ThreadForkResponse = app_server.read_response(id).await?; + app_server.shutdown_gracefully().await?; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + for (method, params) in [ + ("thread/resume", json!({"threadId": child.thread.id})), + ("thread/resume", json!({"threadId": child.thread.id})), + ("thread/fork", json!({"threadId": child.thread.id})), + ] { + let id = app_server.send_raw_request(method, Some(params)).await?; + let _: serde_json::Value = app_server.read_response(id).await?; + } + app_server + .read_stream_until_notification_message("thread/started") + .await?; + assert!( + app_server + .pending_notification_methods() + .iter() + .all(|method| method != "thread/tokenUsage/updated"), + "an empty attached thread must not replay its parent's usage" + ); + Ok(()) } diff --git a/codex-rs/thread-store/src/local/paginated_fork.rs b/codex-rs/thread-store/src/local/paginated_fork.rs index 0b14091267..09a3eede89 100644 --- a/codex-rs/thread-store/src/local/paginated_fork.rs +++ b/codex-rs/thread-store/src/local/paginated_fork.rs @@ -111,7 +111,10 @@ pub(super) async fn history_base_at_boundary( let position = match boundary { ForkBoundary::Latest => latest_position, ForkBoundary::ThroughTurn(turn_id) => { - let row = find_visible_turn(pool, lineage, turn_id.as_str()).await?; + let visible = store + .resolve_rollout_lineage_for_history_page(thread_id) + .await?; + let row = find_visible_turn(pool, &visible, turn_id.as_str()).await?; if row.status == "inProgress" { return Err(ThreadStoreError::InvalidRequest { message: format!("lastTurnId '{turn_id}' identifies an in-progress turn"), @@ -134,7 +137,10 @@ pub(super) async fn history_base_at_boundary( } } ForkBoundary::BeforeTurn(turn_id) => { - let row = find_source_turn(pool, lineage, turn_id.as_str()).await?; + let visible = store + .resolve_rollout_lineage_for_history_page(thread_id) + .await?; + let row = find_source_turn(pool, &visible, turn_id.as_str()).await?; if row.rollout_end_ordinal == Some(row.rollout_ordinal) { return Err(ThreadStoreError::InvalidRequest { message: format!("turn {turn_id} does not have a persisted start boundary"), diff --git a/codex-rs/thread-store/src/local/rollout_lineage.rs b/codex-rs/thread-store/src/local/rollout_lineage.rs index 3008f3751b..fdde832e0a 100644 --- a/codex-rs/thread-store/src/local/rollout_lineage.rs +++ b/codex-rs/thread-store/src/local/rollout_lineage.rs @@ -4,9 +4,12 @@ use std::path::Path; use std::path::PathBuf; use codex_protocol::ThreadId; +use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::ThreadHistoryMode; +use codex_rollout::RolloutItem; + use super::LocalThreadStore; use super::thread_rollout_resolver; use crate::ThreadStoreError; @@ -205,6 +208,20 @@ impl LocalThreadStore { })?, None => 1, }; + // Reverts can point to earlier records of the same child. Stop at the original parent bookmark. + let at_parent = matches!(mode, LineageResolutionMode::HistoryPage) + && meta.meta.attached_thread + && meta.meta.history_base.is_some_and(|base| { + codex_rollout::forked_from_ordinal_exclusive(&meta.meta, Some(&rollout_path)) + .is_some_and(|end| base.end_ordinal_exclusive <= end) + }); + let start_ordinal = if at_parent { + attached_history_start(&rollout_path, start_ordinal) + .await + .map_err(|e| malformed_lineage(requested_thread_id, &e.to_string()))? + } else { + start_ordinal + }; segments.push(RolloutLineageSegment { rollout_id, rollout_path, @@ -212,7 +229,7 @@ impl LocalThreadStore { end, }); - let Some(base) = meta.meta.history_base else { + let Some(base) = meta.meta.history_base.filter(|_| !at_parent) else { break; }; if matches!(mode, LineageResolutionMode::HistoryPage) @@ -235,6 +252,30 @@ impl LocalThreadStore { } } +/// A mid-turn fork starts with its header, settings, optional model guidance, and parent abort. +/// Stop before the first child event; a real child interruption follows its own turn start. +async fn attached_history_start(path: &Path, start: u64) -> std::io::Result { + let mut reader = codex_rollout::open_rollout_line_reader(path).await?; + for _ in 0..4 { + let Some(line) = reader.next_line().await? else { + break; + }; + let Ok(line) = codex_rollout::parse_rollout_line(&line) else { + break; + }; + match line.item { + RolloutItem::SessionMeta(_) + | RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(_)) + | RolloutItem::ResponseItem(_) => {} + RolloutItem::EventMsg(EventMsg::TurnAborted(_)) => { + return Ok(line.ordinal.map_or(start, |n| n.saturating_add(1))); + } + _ => break, + } + } + Ok(start) +} + async fn resolve_rollout_path_by_id( store: &LocalThreadStore, rollout_id: ThreadId, diff --git a/codex-rs/thread-store/src/local/thread_history/realtime.rs b/codex-rs/thread-store/src/local/thread_history/realtime.rs index af0209fd47..cfd3a8ebde 100644 --- a/codex-rs/thread-store/src/local/thread_history/realtime.rs +++ b/codex-rs/thread-store/src/local/thread_history/realtime.rs @@ -53,7 +53,7 @@ pub(in crate::local) async fn list_timeline( validate_page_size(params.page_size)?; let lineage = store - .resolve_rollout_lineage(params.thread_id, /*initial_path*/ None) + .resolve_rollout_lineage_for_history_page(params.thread_id) .await?; let pool = store.thread_history_db().await?; let cursor = params diff --git a/codex-rs/thread-store/src/local/thread_history/search.rs b/codex-rs/thread-store/src/local/thread_history/search.rs index 6aeb6d576a..e3896447a9 100644 --- a/codex-rs/thread-store/src/local/thread_history/search.rs +++ b/codex-rs/thread-store/src/local/thread_history/search.rs @@ -75,7 +75,7 @@ pub(in crate::local) async fn search_thread_occurrences( )?; let matcher = LiteralMatcher::new(params.search_term.as_str()); let lineage = store - .resolve_rollout_lineage(params.thread_id, /*initial_path*/ None) + .resolve_rollout_lineage_for_history_page(params.thread_id) .await?; let cursor_segment = cursor .as_ref()