Keep parent history out of attached thread views (#57256)

## Why

Attached threads need inherited context for model requests while exposing only their own turns to clients. Parent turns and token usage should not appear as activity in an empty attached thread.

## What changed

- Stop attached-thread history traversal at the parent boundary, including after reverts, and hide inherited interruption markers.
- Apply visible-history boundaries to search, realtime timelines, and turn-specific fork and revert requests.
- Suppress inherited token-usage replay on paginated fork resumes, require fork replay to reference a returned turn, and skip empty turn IDs.
- Require a paginated source for `thread/fork` with `attachToSource` and update the protocol documentation.

## Testing

Extend regression coverage for attached-thread restart and database recovery, retained parent context in model requests, child-only history, mid-turn fork boundaries, legacy-source rejection, and suppression of parent token-usage notifications.

GitOrigin-RevId: 67858caf3bb74b7e79d4dd59bed9c61055fe0ff1

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")]

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);
         }

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;
     };

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}),

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"));

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(())
 }
 

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"),

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<u64> {
+    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,

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

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()