Fix fork source display in /status (expose forked_from_id in app server) (#16596)

Addresses #16560 Problem: `/status` stopped showing the source thread id in forked TUI sessions after the app-server migration. Solution: Carry fork source ids through app-server v2 thread data and the TUI session adapter, and update TUI fixtures so `/status` matches the old TUI behavior.
2026-07-01 00:31:56 +08:00 · 2026-04-02 15:05:29 -06:00
parent 93380a6fac
commit 9bb7f0a694
27 changed files with 267 additions and 3 deletions
@@ -58,6 +58,7 @@ use tokio::sync::mpsc;
 fn sample_thread(thread_id: &str, ephemeral: bool) -> Thread {
    Thread {
        id: thread_id.to_string(),
+        forked_from_id: None,
        preview: "first prompt".to_string(),
        ephemeral,
        model_provider: "openai".to_string(),
@@ -2425,6 +2425,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -12186,6 +12186,13 @@
            "description": "Whether the thread is ephemeral and should not be materialized on disk.",
            "type": "boolean"
          },
+          "forkedFromId": {
+            "description": "Source thread id when this thread was created by forking another thread.",
+            "type": [
+              "string",
+              "null"
+            ]
+          },
          "gitInfo": {
            "anyOf": [
              {
@@ -10041,6 +10041,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -1042,6 +1042,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -800,6 +800,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -800,6 +800,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -800,6 +800,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -1042,6 +1042,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -800,6 +800,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -1042,6 +1042,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -800,6 +800,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -800,6 +800,13 @@
          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
          "type": "boolean"
        },
+        "forkedFromId": {
+          "description": "Source thread id when this thread was created by forking another thread.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
        "gitInfo": {
          "anyOf": [
            {
@@ -7,6 +7,10 @@ import type { ThreadStatus } from "./ThreadStatus";
 import type { Turn } from "./Turn";

 export type Thread = { id: string, 
+/**
+ * Source thread id when this thread was created by forking another thread.
+ */
+forkedFromId: string | null, 
 /**
 * Usually the first user message in the thread, if available.
 */
@@ -1307,6 +1307,7 @@ mod tests {
            response: v2::ThreadStartResponse {
                thread: v2::Thread {
                    id: "67e55044-10b1-426f-9247-bb680e5fe0c8".to_string(),
+                    forked_from_id: None,
                    preview: "first prompt".to_string(),
                    ephemeral: true,
                    model_provider: "openai".to_string(),
@@ -1343,6 +1344,7 @@ mod tests {
                "response": {
                    "thread": {
                        "id": "67e55044-10b1-426f-9247-bb680e5fe0c8",
+                        "forkedFromId": null,
                        "preview": "first prompt",
                        "ephemeral": true,
                        "modelProvider": "openai",
@@ -3574,6 +3574,8 @@ impl From<CoreSkillErrorInfo> for SkillErrorInfo {
 #[ts(export_to = "v2/")]
 pub struct Thread {
    pub id: String,
+    /// Source thread id when this thread was created by forking another thread.
+    pub forked_from_id: Option<String>,
    /// Usually the first user message in the thread, if available.
    pub preview: String,
    /// Whether the thread is ephemeral and should not be materialized on disk.
@@ -134,7 +134,7 @@ Example with notification opt-out:

 - `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) 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; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), 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; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. The returned `thread.forkedFromId` points at the source thread when known. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), 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.
@@ -3523,6 +3523,11 @@ impl CodexMessageProcessor {
            }
            build_thread_from_snapshot(thread_uuid, &config_snapshot, loaded_rollout_path)
        };
+        if thread.forked_from_id.is_none()
+            && let Some(rollout_path) = rollout_path.as_ref()
+        {
+            thread.forked_from_id = forked_from_id_from_rollout(rollout_path).await;
+        }
        self.attach_thread_name(thread_uuid, &mut thread).await;

        if include_turns && let Some(rollout_path) = rollout_path.as_ref() {
@@ -4352,7 +4357,12 @@ impl CodexMessageProcessor {
            )
            .await
            {
-                Ok(summary) => summary_to_thread(summary),
+                Ok(summary) => {
+                    let mut thread = summary_to_thread(summary);
+                    thread.forked_from_id =
+                        forked_from_id_from_rollout(fork_rollout_path.as_path()).await;
+                    thread
+                }
                Err(err) => {
                    self.send_internal_error(
                        request_id,
@@ -4386,6 +4396,14 @@ impl CodexMessageProcessor {
                }
            };
            thread.preview = preview_from_rollout_items(&history_items);
+            thread.forked_from_id = source_thread_id
+                .or_else(|| {
+                    history_items.iter().find_map(|item| match item {
+                        RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id),
+                        _ => None,
+                    })
+                })
+                .map(|id| id.to_string());
            if let Err(message) = populate_thread_turns(
                &mut thread,
                ThreadTurnSource::HistoryItems(&history_items),
@@ -8566,6 +8584,7 @@ async fn load_thread_summary_for_rollout(
                rollout_path.display()
            )
        })?;
+    thread.forked_from_id = forked_from_id_from_rollout(rollout_path).await;
    if let Some(persisted_metadata) = persisted_metadata {
        merge_mutable_thread_metadata(
            &mut thread,
@@ -8577,6 +8596,14 @@ async fn load_thread_summary_for_rollout(
    Ok(thread)
 }

+async fn forked_from_id_from_rollout(path: &Path) -> Option<String> {
+    read_session_meta_line(path)
+        .await
+        .ok()
+        .and_then(|meta_line| meta_line.meta.forked_from_id)
+        .map(|thread_id| thread_id.to_string())
+}
+
 fn merge_mutable_thread_metadata(thread: &mut Thread, persisted_thread: Thread) {
    thread.git_info = persisted_thread.git_info;
 }
@@ -8657,6 +8684,7 @@ fn build_thread_from_snapshot(
    let now = time::OffsetDateTime::now_utc().unix_timestamp();
    Thread {
        id: thread_id.to_string(),
+        forked_from_id: None,
        preview: String::new(),
        ephemeral: config_snapshot.ephemeral,
        model_provider: config_snapshot.model_provider_id.clone(),
@@ -8699,6 +8727,7 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread {

    Thread {
        id: conversation_id.to_string(),
+        forked_from_id: None,
        preview,
        ephemeral: false,
        model_provider,
@@ -9195,6 +9224,44 @@ mod tests {
        Ok(())
    }

+    #[tokio::test]
+    async fn read_summary_from_rollout_preserves_forked_from_id() -> Result<()> {
+        use codex_protocol::protocol::RolloutItem;
+        use codex_protocol::protocol::RolloutLine;
+        use codex_protocol::protocol::SessionMetaLine;
+        use std::fs;
+
+        let temp_dir = TempDir::new()?;
+        let path = temp_dir.path().join("rollout.jsonl");
+
+        let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
+        let forked_from_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?;
+        let timestamp = "2025-09-05T16:53:11.850Z".to_string();
+
+        let session_meta = SessionMeta {
+            id: conversation_id,
+            forked_from_id: Some(forked_from_id),
+            timestamp: timestamp.clone(),
+            model_provider: Some("test-provider".to_string()),
+            ..SessionMeta::default()
+        };
+
+        let line = RolloutLine {
+            timestamp,
+            item: RolloutItem::SessionMeta(SessionMetaLine {
+                meta: session_meta,
+                git: None,
+            }),
+        };
+        fs::write(&path, format!("{}\n", serde_json::to_string(&line)?))?;
+
+        assert_eq!(
+            forked_from_id_from_rollout(path.as_path()).await,
+            Some(forked_from_id.to_string())
+        );
+        Ok(())
+    }
+
    #[tokio::test]
    async fn aborting_pending_request_clears_pending_state() -> Result<()> {
        let thread_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?;
@@ -792,6 +792,7 @@ mod tests {
    fn test_thread(thread_id: &str, source: codex_app_server_protocol::SessionSource) -> Thread {
        Thread {
            id: thread_id.to_string(),
+            forked_from_id: None,
            preview: String::new(),
            ephemeral: false,
            model_provider: "mock-provider".to_string(),
@@ -112,6 +112,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
    );

    assert_ne!(thread.id, conversation_id);
+    assert_eq!(thread.forked_from_id, Some(conversation_id.clone()));
    assert_eq!(thread.preview, preview);
    assert_eq!(thread.model_provider, "mock_provider");
    assert_eq!(thread.status, ThreadStatus::Idle);
@@ -7,6 +7,8 @@ use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::SessionSource;
+use codex_app_server_protocol::ThreadForkParams;
+use codex_app_server_protocol::ThreadForkResponse;
 use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadListParams;
 use codex_app_server_protocol::ThreadListResponse;
@@ -152,6 +154,56 @@ async fn thread_read_can_include_turns() -> Result<()> {
    Ok(())
 }

+#[tokio::test]
+async fn thread_read_returns_forked_from_id_for_forked_threads() -> 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_text_elements(
+        codex_home.path(),
+        "2025-01-05T12-00-00",
+        "2025-01-05T12:00:00Z",
+        "Saved user message",
+        vec![],
+        Some("mock_provider"),
+        /*git_info*/ 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.clone(),
+            ..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: forked, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
+
+    let read_id = mcp
+        .send_thread_read_request(ThreadReadParams {
+            thread_id: forked.id,
+            include_turns: false,
+        })
+        .await?;
+    let read_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
+    )
+    .await??;
+    let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
+
+    assert_eq!(thread.forked_from_id, Some(conversation_id));
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn thread_read_loaded_thread_returns_precomputed_path_before_materialization() -> Result<()> {
    let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -242,6 +242,7 @@ async fn resume_lookup_model_providers_filters_only_last_lookup() {
 fn turn_items_for_thread_returns_matching_turn_items() {
    let thread = AppServerThread {
        id: "thread-1".to_string(),
+        forked_from_id: None,
        preview: String::new(),
        ephemeral: false,
        model_provider: "openai".to_string(),
@@ -360,6 +361,7 @@ fn session_configured_from_thread_response_uses_review_policy_from_response() {
    let response = ThreadStartResponse {
        thread: codex_app_server_protocol::Thread {
            id: "67e55044-10b1-426f-9247-bb680e5fe0c8".to_string(),
+            forked_from_id: None,
            preview: String::new(),
            ephemeral: false,
            model_provider: "openai".to_string(),
@@ -8639,6 +8639,7 @@ guardian_approval = true
            ServerNotification::ThreadStarted(ThreadStartedNotification {
                thread: Thread {
                    id: agent_thread_id.to_string(),
+                    forked_from_id: None,
                    preview: "agent thread".to_string(),
                    ephemeral: false,
                    model_provider: "agent-provider".to_string(),
@@ -8719,6 +8720,7 @@ guardian_approval = true
            ServerNotification::ThreadStarted(ThreadStartedNotification {
                thread: Thread {
                    id: agent_thread_id.to_string(),
+                    forked_from_id: None,
                    preview: "agent thread".to_string(),
                    ephemeral: false,
                    model_provider: "agent-provider".to_string(),
@@ -10680,6 +10682,7 @@ guardian_approval = true
            &ThreadRollbackResponse {
                thread: Thread {
                    id: thread_id.to_string(),
+                    forked_from_id: None,
                    preview: String::new(),
                    ephemeral: false,
                    model_provider: "openai".to_string(),
@@ -1251,6 +1251,7 @@ mod tests {
    fn replays_command_execution_items_from_thread_snapshots() {
        let thread = Thread {
            id: "019cee8c-b993-7e33-88c0-014d4e62612d".to_string(),
+            forked_from_id: None,
            preview: String::new(),
            ephemeral: false,
            model_provider: "openai".to_string(),
@@ -1417,6 +1418,7 @@ mod tests {
        let events = thread_snapshot_events(
            &Thread {
                id: thread_id.to_string(),
+                forked_from_id: None,
                preview: "hello".to_string(),
                ephemeral: false,
                model_provider: "openai".to_string(),
@@ -111,6 +111,7 @@ mod tests {
    fn test_thread(thread_id: ThreadId, source: SessionSource) -> Thread {
        Thread {
            id: thread_id.to_string(),
+            forked_from_id: None,
            preview: String::new(),
            ephemeral: false,
            model_provider: "openai".to_string(),
@@ -945,6 +945,7 @@ async fn thread_session_state_from_thread_start_response(
 ) -> Result<ThreadSessionState, String> {
    thread_session_state_from_thread_response(
        &response.thread.id,
+        response.thread.forked_from_id.clone(),
        response.thread.name.clone(),
        response.thread.path.clone(),
        response.model.clone(),
@@ -966,6 +967,7 @@ async fn thread_session_state_from_thread_resume_response(
 ) -> Result<ThreadSessionState, String> {
    thread_session_state_from_thread_response(
        &response.thread.id,
+        response.thread.forked_from_id.clone(),
        response.thread.name.clone(),
        response.thread.path.clone(),
        response.model.clone(),
@@ -987,6 +989,7 @@ async fn thread_session_state_from_thread_fork_response(
 ) -> Result<ThreadSessionState, String> {
    thread_session_state_from_thread_response(
        &response.thread.id,
+        response.thread.forked_from_id.clone(),
        response.thread.name.clone(),
        response.thread.path.clone(),
        response.model.clone(),
@@ -1027,6 +1030,7 @@ fn review_target_to_app_server(
 )]
 async fn thread_session_state_from_thread_response(
    thread_id: &str,
+    forked_from_id: Option<String>,
    thread_name: Option<String>,
    rollout_path: Option<PathBuf>,
    model: String,
@@ -1041,12 +1045,17 @@ async fn thread_session_state_from_thread_response(
 ) -> Result<ThreadSessionState, String> {
    let thread_id = ThreadId::from_string(thread_id)
        .map_err(|err| format!("thread id `{thread_id}` is invalid: {err}"))?;
+    let forked_from_id = forked_from_id
+        .as_deref()
+        .map(ThreadId::from_string)
+        .transpose()
+        .map_err(|err| format!("forked_from_id is invalid: {err}"))?;
    let (history_log_id, history_entry_count) = message_history::history_metadata(config).await;
    let history_entry_count = u64::try_from(history_entry_count).unwrap_or(u64::MAX);

    Ok(ThreadSessionState {
        thread_id,
-        forked_from_id: None,
+        forked_from_id,
        thread_name,
        model,
        model_provider_id,
@@ -1164,9 +1173,11 @@ mod tests {
        let temp_dir = tempfile::tempdir().expect("tempdir");
        let config = build_config(&temp_dir).await;
        let thread_id = ThreadId::new();
+        let forked_from_id = ThreadId::new();
        let response = ThreadResumeResponse {
            thread: codex_app_server_protocol::Thread {
                id: thread_id.to_string(),
+                forked_from_id: Some(forked_from_id.to_string()),
                preview: "hello".to_string(),
                ephemeral: false,
                model_provider: "openai".to_string(),
@@ -1215,6 +1226,7 @@ mod tests {
        let started = started_thread_from_resume_response(response.clone(), &config)
            .await
            .expect("resume response should map");
+        assert_eq!(started.session.forked_from_id, Some(forked_from_id));
        assert_eq!(started.turns.len(), 1);
        assert_eq!(started.turns[0], response.thread.turns[0]);
    }
@@ -1234,6 +1246,7 @@ mod tests {

        let session = thread_session_state_from_thread_response(
            &thread_id.to_string(),
+            /*forked_from_id*/ None,
            Some("restore".to_string()),
            /*rollout_path*/ None,
            "gpt-5.4".to_string(),
@@ -1253,6 +1266,34 @@ mod tests {
        assert_eq!(session.history_entry_count, 2);
    }

+    #[tokio::test]
+    async fn session_configured_preserves_fork_source_thread_id() {
+        let temp_dir = tempfile::tempdir().expect("tempdir");
+        let config = build_config(&temp_dir).await;
+        let thread_id = ThreadId::new();
+        let forked_from_id = ThreadId::new();
+
+        let session = thread_session_state_from_thread_response(
+            &thread_id.to_string(),
+            Some(forked_from_id.to_string()),
+            Some("restore".to_string()),
+            /*rollout_path*/ None,
+            "gpt-5.4".to_string(),
+            "openai".to_string(),
+            /*service_tier*/ None,
+            AskForApproval::Never,
+            codex_protocol::config_types::ApprovalsReviewer::User,
+            SandboxPolicy::new_read_only_policy(),
+            PathBuf::from("/tmp/project"),
+            /*reasoning_effort*/ None,
+            &config,
+        )
+        .await
+        .expect("session should map");
+
+        assert_eq!(session.forked_from_id, Some(forked_from_id));
+    }
+
    #[test]
    fn status_account_display_from_auth_mode_uses_remapped_plan_labels() {
        let business = status_account_display_from_auth_mode(
@@ -2639,6 +2639,7 @@ mod tests {
        let thread_id = ThreadId::new();
        let thread = Thread {
            id: thread_id.to_string(),
+            forked_from_id: None,
            preview: String::from("remote thread"),
            ephemeral: false,
            model_provider: String::from("openai"),