Use paginated history for persistent exec threads (#38774)

## What changed

- Request paginated history when `codex exec` starts a persistent thread.
- Keep ephemeral threads unchanged and retry with legacy history when the configured thread store does not support pagination.
- Exercise paginated resume and fork persistence, including resumed history and fork ancestry without copying the source transcript.

## Testing

- Extend unit coverage for persistent and ephemeral thread start parameters.
- Add an integration test for the legacy-history fallback.

GitOrigin-RevId: 610a2db14524b127551e75b7aef541d44368f28a

codex-rs/exec/src/lib.rs

@@ -19,6 +19,7 @@ use codex_app_server_client::ExecServerRuntimePaths;
 use codex_app_server_client::InProcessAppServerClient;
 use codex_app_server_client::InProcessClientStartArgs;
 use codex_app_server_client::InProcessServerEvent;
+use codex_app_server_client::TypedRequestError;
 use codex_app_server_protocol::ClientRequest;
 use codex_app_server_protocol::ConfigWarningNotification;
 use codex_app_server_protocol::JSONRPCErrorError;
@@ -33,6 +34,7 @@ use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::Thread as AppServerThread;
 use codex_app_server_protocol::ThreadForkParams;
 use codex_app_server_protocol::ThreadForkResponse;
+use codex_app_server_protocol::ThreadHistoryMode;
 use codex_app_server_protocol::ThreadItem as AppServerThreadItem;
 use codex_app_server_protocol::ThreadListParams;
 use codex_app_server_protocol::ThreadListResponse;
@@ -834,16 +836,9 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
                     .map_err(anyhow::Error::msg)?;
             (session_configured.thread_id, session_configured)
         } else {
-            let response: ThreadStartResponse = send_request_with_response(
-                &client,
-                ClientRequest::ThreadStart {
-                    request_id: request_ids.next(),
-                    params: thread_start_params_from_config(&config),
-                },
-                "thread/start",
-            )
-            .await
-            .map_err(anyhow::Error::msg)?;
+            let response = start_thread(&client, &mut request_ids, &config)
+                .await
+                .map_err(anyhow::Error::msg)?;
             let session_configured =
                 session_configured_from_thread_start_response(&response, &config)
                     .map_err(anyhow::Error::msg)?;
@@ -915,16 +910,9 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
         .map_err(anyhow::Error::msg)?;
         (session_configured.thread_id, session_configured)
     } else {
-        let response: ThreadStartResponse = send_request_with_response(
-            &client,
-            ClientRequest::ThreadStart {
-                request_id: request_ids.next(),
-                params: thread_start_params_from_config(&config),
-            },
-            "thread/start",
-        )
-        .await
-        .map_err(anyhow::Error::msg)?;
+        let response = start_thread(&client, &mut request_ids, &config)
+            .await
+            .map_err(anyhow::Error::msg)?;
         let session_configured = session_configured_from_thread_start_response(&response, &config)
             .map_err(anyhow::Error::msg)?;
         (session_configured.thread_id, session_configured)
@@ -1148,6 +1136,34 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
     Ok(())
 }
 
+async fn start_thread(
+    client: &InProcessAppServerClient,
+    request_ids: &mut RequestIdSequencer,
+    config: &Config,
+) -> Result<ThreadStartResponse, String> {
+    let mut params = thread_start_params_from_config(config);
+    loop {
+        match client
+            .request_typed(ClientRequest::ThreadStart {
+                request_id: request_ids.next(),
+                params: params.clone(),
+            })
+            .await
+        {
+            Ok(response) => return Ok(response),
+            Err(TypedRequestError::Server { source, .. })
+                if params.history_mode.is_some()
+                    && source.code == -32600
+                    && source.message
+                        == "paginated threads require thread/turns/list and thread/items/list support" =>
+            {
+                params.history_mode = None;
+            }
+            Err(err) => return Err(format!("thread/start: {err}")),
+        }
+    }
+}
+
 fn thread_start_params_from_config(config: &Config) -> ThreadStartParams {
     let permissions = permissions_selection_from_config(config);
     let sandbox = permissions.is_none().then(|| {
@@ -1167,6 +1183,7 @@ fn thread_start_params_from_config(config: &Config) -> ThreadStartParams {
         permissions,
         config: thread_config_overrides_from_config(config),
         ephemeral: Some(config.ephemeral),
+        history_mode: (!config.ephemeral).then_some(ThreadHistoryMode::Paginated),
         thread_source: Some(ThreadSource::User),
         ..ThreadStartParams::default()
     }

codex-rs/exec/src/lib_tests.rs

@@ -613,10 +613,10 @@ async fn build_exec_config_preserves_headless_error_when_retry_fails() {
 }
 
 #[tokio::test]
-async fn thread_start_params_include_user_thread_source() {
+async fn thread_start_params_match_history_to_persistence() {
     let codex_home = tempdir().expect("create temp codex home");
     let cwd = tempdir().expect("create temp cwd");
-    let config = ConfigBuilder::default()
+    let mut config = ConfigBuilder::default()
         .codex_home(codex_home.path().to_path_buf())
         .fallback_cwd(Some(cwd.path().to_path_buf()))
         .build()
@@ -629,6 +629,13 @@ async fn thread_start_params_include_user_thread_source() {
         params.thread_source,
         Some(codex_app_server_protocol::ThreadSource::User)
     );
+    assert_eq!(params.history_mode, Some(ThreadHistoryMode::Paginated));
+
+    config.ephemeral = true;
+    let params = thread_start_params_from_config(&config);
+
+    assert_eq!(params.ephemeral, Some(true));
+    assert_eq!(params.history_mode, None);
 }
 
 #[tokio::test]

codex-rs/exec/tests/suite/resume.rs

@@ -132,13 +132,36 @@ async fn mount_exec_responses(
     responses::mount_sse_sequence(server, (0..count).map(exec_sse_response).collect()).await
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_falls_back_to_legacy_history_when_thread_store_cannot_paginate() -> anyhow::Result<()>
+{
+    skip_if_no_network!(Ok(()));
+
+    let test = test_codex_exec();
+    let server = MockServer::start().await;
+    let _response_mock = mount_exec_responses(&server, /*count*/ 1).await;
+    let store_id = Uuid::new_v4();
+
+    test.cmd_with_server(&server)
+        .arg("--skip-git-repo-check")
+        .arg("-c")
+        .arg(format!(
+            "experimental_thread_store={{type=\"in_memory\",id=\"{store_id}\"}}"
+        ))
+        .arg("continue without paginated history")
+        .assert()
+        .success();
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn exec_resume_last_appends_to_existing_file() -> anyhow::Result<()> {
     skip_if_no_network!(Ok(()));
 
     let test = test_codex_exec();
     let server = MockServer::start().await;
-    let _response_mock = responses::mount_sse_sequence(
+    let response_mock = responses::mount_sse_sequence(
         &server,
         vec![
             responses::sse(vec![
@@ -168,6 +191,14 @@ async fn exec_resume_last_appends_to_existing_file() -> anyhow::Result<()> {
     let sessions_dir = test.home_path().join("sessions");
     let path = find_session_file_containing_marker(&sessions_dir, &marker)
         .expect("no session file found after first run");
+    let content = std::fs::read_to_string(&path)?;
+    let meta: Value = serde_json::from_str(
+        content
+            .lines()
+            .next()
+            .expect("rollout should contain session metadata"),
+    )?;
+    assert_eq!(meta["payload"]["history_mode"], "paginated");
 
     // 2) Second run: resume the most recent file with a new marker.
     let marker2 = format!("resume-last-2-{}", Uuid::new_v4());
@@ -190,8 +221,8 @@ async fn exec_resume_last_appends_to_existing_file() -> anyhow::Result<()> {
         stderr
             .matches("app-server event: thread/tokenUsage/updated")
             .count(),
-        1,
-        "resume should not replay restored token usage: {stderr}"
+        2,
+        "paginated resume should replay restored token usage before the new turn: {stderr}"
     );
 
     // Ensure the same file was updated and contains both markers.
@@ -204,6 +235,11 @@ async fn exec_resume_last_appends_to_existing_file() -> anyhow::Result<()> {
     let content = std::fs::read_to_string(&resumed_path)?;
     assert!(content.contains(&marker));
     assert!(content.contains(&marker2));
+    let requests = response_mock.requests();
+    assert_eq!(requests.len(), 2);
+    let resumed_request = requests[1].body_json().to_string();
+    assert!(resumed_request.contains(&marker));
+    assert!(resumed_request.contains(&marker2));
     Ok(())
 }
 
@@ -982,7 +1018,8 @@ async fn exec_fork_creates_distinct_threads_with_and_without_a_prompt() -> anyho
             .expect("fork rollout should contain session metadata"),
     )?;
     assert_eq!(fork_meta["payload"]["forked_from_id"], source_id);
-    assert!(fork_contents.contains(&source_marker));
+    assert_eq!(fork_meta["payload"]["history_base"]["thread_id"], source_id);
+    assert!(!fork_contents.contains(&source_marker));
     assert!(fork_contents.contains(&fork_marker));
     assert_eq!(std::fs::read_to_string(&source_path)?, original_source);