openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support ephemeral forks of paginated threads (#35251)

Browse Source 
## What changed

- Allow `thread/fork` to create an ephemeral fork from paginated history when
  `excludeTurns: true` is set, and return an invalid-request error otherwise.
- Preserve the selected history and fork preview without creating a rollout
  path or exposing the ephemeral thread through `thread/list`.

## Testing

- Cover full-history and bounded ephemeral forks, active-turn boundaries,
  preview selection, model input, and thread listing behavior.

GitOrigin-RevId: 4c0d9739b3310d7d0d7389bec127045ac4f55b44
This commit is contained in:
Owen Lin
2026-07-24 21:58:52 +00:00
committed by copyberry
parent 89a3b89c4c
commit 1811b67a84
3 changed files with 168 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/README.md
+1 -1
View File
@@ -378,7 +378,7 @@ To branch from a stored session, call `thread/fork` with the `thread.id`. This c
{ "method": "thread/started", "params": { "thread": { … } } }
```
Like `thread/resume`, experimental clients can pass `excludeTurns: true` to `thread/fork` to return only thread metadata in `thread.turns` and page history with `thread/turns/list`. In that mode the server skips replaying restored `thread/tokenUsage/updated`, which keeps the fork path from rebuilding turns just to attribute historical usage.
Like `thread/resume`, experimental clients can pass `excludeTurns: true` to `thread/fork` to return only thread metadata in `thread.turns` and page history with `thread/turns/list`. In that mode the server skips replaying restored `thread/tokenUsage/updated`, which keeps the fork path from rebuilding turns just to attribute historical usage. Ephemeral forks of paginated threads require `excludeTurns: true`.
### Example: List threads (with pagination & filters)
codex-rs/app-server/src/request_processors/thread_processor.rs
+11 -6
View File
@@ -4010,17 +4010,17 @@ impl ThreadRequestProcessor {
"`deferGoalContinuation` cannot be combined with `ephemeral`",
));
}
if paginated_source && ephemeral && include_turns {
return Err(invalid_request(
"ephemeral paginated thread/fork requires `excludeTurns: true`",
));
}
let source_thread_id = source_thread.thread_id;
let source_thread_name = source_thread
.name
.as_deref()
.and_then(codex_core::util::normalize_thread_name);
let prepared_fork = if paginated_source {
if ephemeral {
return Err(invalid_request(
"ephemeral paginated thread/fork is not supported",
));
}
let boundary = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
(Some(turn_id), None) => {
codex_thread_store::ForkBoundary::ThroughTurn(turn_id.to_string())
@@ -4301,7 +4301,12 @@ impl ThreadRequestProcessor {
&config_snapshot,
/*path*/ None,
);
thread.preview = preview_from_rollout_items(&history_items);
thread.preview =
if paginated_source && last_turn_id.is_none() && before_turn_id.is_none() {
source_thread.preview.clone()
} else {
preview_from_rollout_items(&history_items)
};
thread.forked_from_id = Some(source_thread_id.to_string());
if include_turns {
populate_thread_turns_from_history(
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+156 -16
View File
@@ -565,6 +565,35 @@ async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix(
.collect::<Vec<_>>(),
turn_ids[..2]
);
let completed = timeout(
DEFAULT_READ_TIMEOUT,
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: forked_thread.id.clone(),
input: vec![UserInput::Text {
text: "private child prompt".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
let ThreadForkResponse {
thread: ephemeral_fork,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: forked_thread.id,
before_turn_id: Some(completed.turn.id),
ephemeral: true,
exclude_turns: true,
..Default::default()
},
})
.await?;
assert_eq!(ephemeral_fork.preview, "first");
}
Ok(())
@@ -1392,6 +1421,22 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted(
.build_initialized()
.await?;
let ThreadForkResponse {
thread: ephemeral_fork,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: source_thread_id.clone(),
thread_source: thread_source.clone(),
ephemeral: true,
exclude_turns: true,
..Default::default()
},
})
.await?;
let invalid_fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id.clone(),
@@ -1470,6 +1515,48 @@ async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted(
)
.await?;
let _: TurnStartResponse = mcp
.request(|request_id| ClientRequest::TurnStart {
request_id,
params: TurnStartParams {
thread_id: ephemeral_fork.id,
input: vec![UserInput::Text {
text: "Continue in an ephemeral fork".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = server.received_requests().await.expect("response requests");
let input = requests
.iter()
.rev()
.find(|request| request.url.path().ends_with("/responses"))
.expect("ephemeral fork model request")
.body_json::<Value>()?["input"]
.clone();
let serialized_input = serde_json::to_string(&input)?;
assert!(serialized_input.contains("before-fork model input"));
assert!(!serialized_input.contains("after-fork model input"));
assert!(input.as_array().is_some_and(|items| {
items.iter().any(|item| {
item["role"] == expected_marker_role
&& item["content"].as_array().is_some_and(|content| {
content.iter().any(|fragment| {
fragment["text"]
.as_str()
.is_some_and(|text| text.contains("<turn_aborted>"))
})
})
})
}));
let ThreadTurnsListResponse { data: turns, .. } = mcp
.request(|request_id| ClientRequest::ThreadTurnsList {
request_id,
@@ -1767,12 +1854,28 @@ async fn thread_fork_surfaces_cloud_config_bundle_load_errors() -> Result<()> {
#[tokio::test]
async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()> {
assert_thread_fork_ephemeral_remains_pathless_and_omits_listing(ThreadHistoryMode::Legacy).await
}
#[tokio::test]
async fn paginated_thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()> {
assert_thread_fork_ephemeral_remains_pathless_and_omits_listing(ThreadHistoryMode::Paginated)
.await
}
async fn assert_thread_fork_ephemeral_remains_pathless_and_omits_listing(
history_mode: ThreadHistoryMode,
) -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let preview = "Saved user message";
let conversation_id = create_fake_rollout(
let create_rollout = match history_mode {
ThreadHistoryMode::Legacy => create_fake_rollout,
ThreadHistoryMode::Paginated => create_fake_paginated_rollout,
};
let conversation_id = create_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
@@ -1787,10 +1890,30 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()
.build_initialized()
.await?;
if history_mode == ThreadHistoryMode::Paginated {
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
ephemeral: true,
..Default::default()
})
.await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert_eq!(
error.error.message,
"ephemeral paginated thread/fork requires `excludeTurns: true`"
);
}
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
ephemeral: true,
exclude_turns: history_mode == ThreadHistoryMode::Paginated,
..Default::default()
})
.await?;
@@ -1814,22 +1937,26 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()
assert_eq!(thread.preview, preview);
assert_eq!(thread.status, ThreadStatus::Idle);
assert_eq!(thread.name, None);
assert_eq!(thread.turns.len(), 1, "expected copied fork history");
if history_mode == ThreadHistoryMode::Paginated {
assert!(thread.turns.is_empty());
} else {
assert_eq!(thread.turns.len(), 1, "expected copied fork history");
let turn = &thread.turns[0];
assert_eq!(turn.status, TurnStatus::Completed);
assert_eq!(turn.items.len(), 1, "expected user message item");
match &turn.items[0] {
ThreadItem::UserMessage { content, .. } => {
assert_eq!(
content,
&vec![UserInput::Text {
text: preview.to_string(),
text_elements: Vec::new(),
}]
);
let turn = &thread.turns[0];
assert_eq!(turn.status, TurnStatus::Completed);
assert_eq!(turn.items.len(), 1, "expected user message item");
match &turn.items[0] {
ThreadItem::UserMessage { content, .. } => {
assert_eq!(
content,
&vec![UserInput::Text {
text: preview.to_string(),
text_elements: Vec::new(),
}]
);
}
other => panic!("expected user message item, got {other:?}"),
}
other => panic!("expected user message item, got {other:?}"),
}
let thread_json = fork_result
@@ -1898,7 +2025,7 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: fork_thread_id,
thread_id: fork_thread_id.clone(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: "continue".to_string(),
@@ -1914,6 +2041,19 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()
)
.await??;
let requests = server.received_requests().await.expect("response requests");
let model_input = requests
.iter()
.find(|request| request.url.path().ends_with("/responses"))
.expect("ephemeral fork model request")
.body_json::<Value>()?["input"]
.to_string();
assert!(model_input.contains(preview));
assert!(model_input.contains("continue"));
let ThreadListResponse { data, .. } = list_threads(&mut mcp).await?;
assert!(data.iter().all(|thread| thread.id != fork_thread_id));
Ok(())
}