Reuse unchanged history snapshots when resuming threads (#49600)

## Why

Resuming a thread rereads persisted history even when a preloaded snapshot is unchanged. Reusing that snapshot must still account for writes committed before writer ownership is acquired.

## What changed

- Carry optional, non-serialized revisions from stored history and model context through thread and agent resume paths.
- Validate the revision under writer ownership and reuse the shared snapshot when it matches. Reload stored history when the revision is missing or changed, while preserving explicit history overrides.
- Resolve the selected rollout before loading context, validate compressed snapshots before materialization, and enforce archived-thread restrictions on canonical paths.

## Testing

Extend coverage for snapshot reuse and late-write refresh in legacy and paginated histories, compressed rollout reuse, archived rollout symlinks, and cold child restoration that includes edits made after preloading history.

GitOrigin-RevId: e1fd7aee6ef2b30f41951e6b7e6b01c3e1420cab

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

@@ -4598,6 +4598,7 @@ impl ThreadRequestProcessor {
                 .await
                 .map_err(thread_store_resume_read_error)?;
             let history = InitialHistory::Resumed(ResumedHistory {
+                history_revision: model_context.revision,
                 conversation_id: model_context.thread_id,
                 history: Arc::new(model_context.items),
                 rollout_path: stored_thread.rollout_path.clone(),
@@ -4692,18 +4693,15 @@ impl ThreadRequestProcessor {
         stored_thread: &mut StoredThread,
     ) -> Result<InitialHistory, JSONRPCErrorError> {
         let thread_id = stored_thread.thread_id;
-        let history = stored_thread
-            .history
-            .take()
-            .map(|history| history.items)
-            .ok_or_else(|| {
-                internal_error(format!(
-                    "thread {thread_id} did not include persisted history"
-                ))
-            })?;
+        let history = stored_thread.history.take().ok_or_else(|| {
+            internal_error(format!(
+                "thread {thread_id} did not include persisted history"
+            ))
+        })?;
         Ok(InitialHistory::Resumed(ResumedHistory {
+            history_revision: history.revision,
             conversation_id: thread_id,
-            history: Arc::new(history),
+            history: Arc::new(history.items),
             rollout_path: stored_thread.rollout_path.clone(),
         }))
     }
@@ -5076,6 +5074,7 @@ impl ThreadRequestProcessor {
         // The fork cutoff can remove the only TurnContext that records the selected version.
         // Recover it from the untrimmed source or live parent, independently of permission overrides.
         let source_multi_agent_version = InitialHistory::Resumed(ResumedHistory {
+            history_revision: None,
             conversation_id: source_thread_id,
             history: Arc::clone(latest_context.as_ref().unwrap_or(&source_history_items)),
             rollout_path: source_thread.rollout_path.clone(),
@@ -5203,6 +5202,7 @@ impl ThreadRequestProcessor {
                     ForkSnapshot::Interrupted,
                     fork_options,
                     InitialHistory::Resumed(ResumedHistory {
+                        history_revision: None,
                         conversation_id: source_thread_id,
                         history: history_items,
                         rollout_path: source_thread.rollout_path.clone(),