Preserve thread context when retrying or editing turns

codex-rs/app-server-protocol/src/protocol/v2/thread.rs

@@ -501,6 +501,12 @@ pub struct ThreadForkParams {
     #[ts(optional = nullable)]
     pub last_turn_id: Option<String>,
 
+    /// Optional turn id to fork before, excluding that turn and all later turns.
+    /// Cannot be combined with `last_turn_id`.
+    #[experimental("thread/fork.beforeTurnId")]
+    #[ts(optional = nullable)]
+    pub before_turn_id: Option<String>,
+
     /// [UNSTABLE] Specify the rollout path to fork from.
     /// If specified, the thread_id param will be ignored.
     #[experimental("thread/fork.path")]
@@ -561,6 +567,12 @@ pub struct ThreadForkParams {
     #[experimental("thread/fork.excludeTurns")]
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub exclude_turns: bool,
+    /// When true, carry the source thread's current goal into the fork without
+    /// starting its initial automatic continuation. The next explicit turn owns
+    /// the goal lifecycle, and normal automatic continuation resumes after it.
+    #[experimental("thread/fork.deferGoalContinuation")]
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub defer_goal_continuation: bool,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)]

codex-rs/app-server/README.md

@@ -139,7 +139,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. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `allowProviderModelFallback` lets providers backed by an authoritative static model catalog replace an unavailable requested `model` with the catalog default; dynamic or cached catalogs preserve the requested model. Experimental `runtimeWorkspaceRoots` supplies the runtime workspace roots used when app-server creates default environment selections; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Deprecated experimental `multiAgentMode` is ignored; use Ultra reasoning effort for proactive multi-agent behavior. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd` and optional environment-native `runtimeWorkspaceRoots`. Explicit environments ignore the top-level roots; omitted per-environment roots default to that environment's `cwd`, while an empty list explicitly selects no roots. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots using environment-native absolute paths. Skills found below those roots are listed and read through the owning environment. Stdio MCP servers declared by selected plugins are started in that environment, and HTTP MCP connections use that environment's HTTP client.
 - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Accepts the same permission override rules as `thread/start`.
-- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; pass an optional `lastTurnId` to copy history only through that turn, inclusive, and drop later turns from the fork. An in-progress `lastTurnId` is rejected. If `lastTurnId` is null while the source thread is 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. Experimental clients can pass `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately. Accepts the same permission override rules as `thread/start`.
+- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; pass an optional `lastTurnId` to copy history only through that turn, inclusive, and drop later turns from the fork. An in-progress `lastTurnId` boundary is rejected. Experimental `beforeTurnId` instead copies history strictly before the referenced turn, including when that turn is in progress, and cannot be combined with `lastTurnId`. If both boundaries are null while the source thread is 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. Experimental clients can pass `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately, or `deferGoalContinuation: true` to carry the source thread's current goal into the fork and run an explicit turn before automatic continuation resumes. Deferred goal continuation is persisted until that turn starts and cannot be combined with `ephemeral: true`. Accepts the same permission override rules as `thread/start`.
 - `thread/start`, `thread/resume`, and `thread/fork` responses include the legacy `sandbox` compatibility projection. `instructionSources` lists loaded instruction files using each source environment's native absolute path syntax, including files loaded from remote environments. Experimental clients can read `runtimeWorkspaceRoots` for the thread-scoped runtime roots and `activePermissionProfile` for the named or implicit built-in profile identity/provenance when known. Their deprecated experimental `multiAgentMode` field, and the corresponding thread setting, always report `explicitRequestOnly`; Ultra reasoning effort is the source of proactive multi-agent behavior.
 - `thread/list` — page through stored threads; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Experimental clients can use `parentThreadId` for direct spawned children or `ancestorThreadId` for spawned descendants at any depth; the two filters are mutually exclusive. Review and Guardian threads are not included because they do not participate in that spawn-edge lifecycle. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. Subagent threads also include `parentThreadId` when the immediate parent is known.
 - `thread/loaded/list` — list the thread ids currently loaded in memory.

codex-rs/app-server/src/request_processors.rs

@@ -343,6 +343,7 @@ use codex_core::path_utils;
 use codex_core::read_head_for_summary;
 use codex_core::sandboxing::SandboxPermissions;
 use codex_core::truncate_rollout_after_turn_id;
+use codex_core::truncate_rollout_before_turn_id;
 use codex_core::windows_sandbox::WindowsSandboxLevelExt;
 use codex_core::windows_sandbox::WindowsSandboxSetupMode as CoreWindowsSandboxSetupMode;
 use codex_core::windows_sandbox::WindowsSandboxSetupRequest;
@@ -523,6 +524,7 @@ mod plugins;
 mod process_exec_processor;
 mod remote_control_processor;
 mod search;
+mod thread_fork_goal;
 mod thread_processor;
 mod token_usage_replay;
 mod turn_processor;

codex-rs/app-server/src/request_processors/thread_fork_goal.rs

@@ -0,0 +1,28 @@
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::validate_thread_goal_objective;
+use codex_state::StateRuntime;
+
+pub(super) async fn inherit_thread_goal_snapshot(
+    state_db: &StateRuntime,
+    source_thread_id: ThreadId,
+    target_thread_id: ThreadId,
+) -> anyhow::Result<bool> {
+    let Some(mut goal) = state_db
+        .thread_goals()
+        .get_thread_goal(source_thread_id)
+        .await?
+    else {
+        return Ok(false);
+    };
+    if let Err(err) = validate_thread_goal_objective(&goal.objective) {
+        tracing::warn!(%source_thread_id, "skipping invalid inherited thread goal: {err}");
+        return Ok(false);
+    }
+
+    goal.thread_id = target_thread_id;
+    state_db
+        .thread_goals()
+        .replace_thread_goal_snapshot(&goal)
+        .await?;
+    Ok(true)
+}

codex-rs/app-server/src/request_processors/thread_goal_processor.rs

@@ -94,6 +94,26 @@ impl ThreadGoalRequestProcessor {
         (emit_thread_goal_update, thread_goal_state_db)
     }
 
+    pub(crate) async fn restore_inherited_goal_runtime(&self, thread_id: ThreadId) {
+        if let Err(err) = self
+            .goal_service
+            .restore_thread_runtime_after_resume(thread_id)
+            .await
+        {
+            warn!("failed to restore inherited goal runtime for {thread_id}: {err}");
+        }
+    }
+
+    pub(crate) async fn flush_goal_progress_for_fork(
+        &self,
+        thread_id: ThreadId,
+    ) -> Result<(), String> {
+        self.goal_service
+            .flush_thread_goal_progress_for_fork(thread_id)
+            .await
+            .map_err(|err| err.to_string())
+    }
+
     async fn thread_goal_set_inner(
         &self,
         request_id: ConnectionRequestId,
@@ -283,7 +303,7 @@ impl ThreadGoalRequestProcessor {
         Ok(())
     }
 
-    async fn emit_thread_goal_snapshot(&self, thread_id: ThreadId) {
+    pub(crate) async fn emit_thread_goal_snapshot(&self, thread_id: ThreadId) {
         let state_db = match self.state_db_for_materialized_thread(thread_id).await {
             Ok(state_db) => state_db,
             Err(err) => {

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -1,3 +1,4 @@
+use super::thread_fork_goal::inherit_thread_goal_snapshot;
 use super::*;
 use crate::error_code::method_not_found;
 use codex_app_server_protocol::SelectedCapabilityRoot;
@@ -3549,6 +3550,7 @@ impl ThreadRequestProcessor {
         let ThreadForkParams {
             thread_id,
             last_turn_id,
+            before_turn_id,
             path,
             model,
             model_provider,
@@ -3565,6 +3567,7 @@ impl ThreadRequestProcessor {
             ephemeral,
             thread_source,
             exclude_turns,
+            defer_goal_continuation,
         } = params;
         let include_turns = !exclude_turns;
         if sandbox.is_some() && permissions.is_some() {
@@ -3582,6 +3585,16 @@ impl ThreadRequestProcessor {
         if matches!(source_thread.history_mode, ThreadHistoryMode::Paginated) {
             return Err(method_not_found("paginated_threads is not supported yet"));
         }
+        if last_turn_id.is_some() && before_turn_id.is_some() {
+            return Err(invalid_request(
+                "`beforeTurnId` cannot be combined with `lastTurnId`",
+            ));
+        }
+        if ephemeral && defer_goal_continuation {
+            return Err(invalid_request(
+                "`deferGoalContinuation` cannot be combined with `ephemeral`",
+            ));
+        }
         let mut source_thread = self
             .read_stored_thread_for_resume(&thread_id, path.as_ref(), /*include_history*/ true)
             .await?;
@@ -3599,13 +3612,17 @@ impl ThreadRequestProcessor {
                     "thread {source_thread_id} did not include persisted history"
                 ))
             })?;
-        let history_items = if let Some(last_turn_id) = last_turn_id.as_deref() {
-            Arc::new(
+        let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
+            (Some(last_turn_id), None) => Arc::new(
                 truncate_rollout_after_turn_id(&history_items, last_turn_id)
                     .map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
-            )
-        } else {
-            Arc::new(history_items)
+            ),
+            (None, Some(before_turn_id)) => Arc::new(
+                truncate_rollout_before_turn_id(&history_items, before_turn_id)
+                    .map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
+            ),
+            (None, None) => Arc::new(history_items),
+            (Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"),
         };
         let history_cwd = Some(source_thread.cwd.clone());
 
@@ -3653,6 +3670,7 @@ impl ThreadRequestProcessor {
             .load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
             .await
             .map_err(|err| config_load_error(&err))?;
+        let goals_enabled = config.features.enabled(Feature::Goals);
 
         let fallback_model_provider = config.model_provider_id.clone();
 
@@ -3705,6 +3723,33 @@ impl ThreadRequestProcessor {
                 .await
                 .map_err(|err| core_thread_write_error("inherit source thread name", err))?;
         }
+        let inherited_goal = if defer_goal_continuation
+            && session_configured.rollout_path.is_some()
+            && goals_enabled
+        {
+            if let Some(state_db) = forked_thread.state_db().or_else(|| self.state_db.clone()) {
+                self.thread_goal_processor
+                    .flush_goal_progress_for_fork(source_thread_id)
+                    .await
+                    .map_err(|err| {
+                        internal_error(format!("failed to flush source thread goal: {err}"))
+                    })?;
+                inherit_thread_goal_snapshot(&state_db, source_thread_id, thread_id)
+                    .await
+                    .map_err(|err| {
+                        internal_error(format!("failed to inherit source thread goal: {err}"))
+                    })?
+            } else {
+                false
+            }
+        } else {
+            false
+        };
+        if inherited_goal {
+            self.thread_goal_processor
+                .restore_inherited_goal_runtime(thread_id)
+                .await;
+        }
 
         let instruction_sources = forked_thread.legacy_instruction_sources().await;
 
@@ -3822,6 +3867,11 @@ impl ThreadRequestProcessor {
         self.outgoing
             .send_server_notification(ServerNotification::ThreadStarted(notif))
             .await;
+        if inherited_goal {
+            self.thread_goal_processor
+                .emit_thread_goal_snapshot(thread_id)
+                .await;
+        }
         Ok(())
     }