Reject directory rollout paths for pathless side chats (#25661)

## Why

Fixes openai/codex#20944.

Desktop side chats are intentionally ephemeral and pathless. They can
still accept live turns while loaded, but after a reload there is no
persisted rollout to resume. In the reported failure mode, Desktop could
send `$CODEX_HOME` as the resume/fork path for one of these pathless
side chats.

`thread/resume` and `thread/fork` prefer an explicit `path` over
`threadId`, and rollout path lookup only checked that a candidate
existed. That let `$CODEX_HOME` pass as a rollout path, so the later
rollout reader tried to open a directory and surfaced the low-level `Is
a directory` error.

## What Changed

- Reject explicit rollout paths that resolve to a directory or other
non-file before attempting to read rollout history.
- Make `codex_rollout::existing_rollout_path` return only plain or
compressed rollout candidates that are actual files.
- Add an app-server regression test that creates an ephemeral fork, runs
a turn while the side thread is loaded, simulates reload, then verifies
both `thread/resume` and `thread/fork` reject `$CODEX_HOME` with `path
is a directory` instead of the OS-level directory-read error.
- Rebase over the `TestAppServer` rename and update the remaining stale
test harness call sites to use `TestAppServer` with `app_server` local
variables.

Relevant code:

- `thread-store/src/local/read_thread.rs` validates explicit rollout
paths before rollout reading:
25b47c8f42/codex-rs/thread-store/src/local/read_thread.rs (L146-L165)
- `rollout/src/compression.rs` now requires file metadata for plain and
compressed rollout candidates:
25b47c8f42/codex-rs/rollout/src/compression.rs (L940-L950)
- The repro test covers the pathless ephemeral side-chat reload case:
25b47c8f42/codex-rs/app-server/tests/suite/v2/thread_fork.rs (L774-L886)

## Verification

- `just test -p codex-app-server
pathless_ephemeral_thread_rejects_codex_home_path_after_reload`

codex-rs/app-server/tests/suite/v2/thread_fork.rs

@@ -17,6 +17,7 @@ use codex_app_server_protocol::ThreadForkResponse;
 use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadListParams;
 use codex_app_server_protocol::ThreadListResponse;
+use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadSource;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
@@ -770,6 +771,120 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()
     Ok(())
 }
 
+#[tokio::test]
+async fn pathless_ephemeral_thread_rejects_codex_home_path_after_reload() -> 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 parent_thread_id = create_fake_rollout(
+        codex_home.path(),
+        "2025-01-05T12-00-00",
+        "2025-01-05T12:00:00Z",
+        "Parent message",
+        Some("mock_provider"),
+        /*git_info*/ None,
+    )?;
+
+    let side_thread_id = {
+        let mut app_server = TestAppServer::new(codex_home.path()).await?;
+        timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??;
+
+        let fork_id = app_server
+            .send_thread_fork_request(ThreadForkParams {
+                thread_id: parent_thread_id,
+                ephemeral: true,
+                ..Default::default()
+            })
+            .await?;
+        let fork_resp: JSONRPCResponse = timeout(
+            DEFAULT_READ_TIMEOUT,
+            app_server.read_stream_until_response_message(RequestId::Integer(fork_id)),
+        )
+        .await??;
+        let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
+        assert!(thread.ephemeral);
+        assert_eq!(thread.path, None);
+
+        let turn_id = app_server
+            .send_turn_start_request(TurnStartParams {
+                thread_id: thread.id.clone(),
+                client_user_message_id: None,
+                input: vec![UserInput::Text {
+                    text: "continue".to_string(),
+                    text_elements: Vec::new(),
+                }],
+                ..Default::default()
+            })
+            .await?;
+        let turn_resp: JSONRPCResponse = timeout(
+            DEFAULT_READ_TIMEOUT,
+            app_server.read_stream_until_response_message(RequestId::Integer(turn_id)),
+        )
+        .await??;
+        let _: TurnStartResponse = to_response::<TurnStartResponse>(turn_resp)?;
+        timeout(
+            DEFAULT_READ_TIMEOUT,
+            app_server.read_stream_until_notification_message("turn/completed"),
+        )
+        .await??;
+
+        thread.id
+    };
+
+    let mut app_server = TestAppServer::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??;
+    let codex_home_path = codex_home.path().to_path_buf();
+
+    let resume_id = app_server
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: side_thread_id.clone(),
+            path: Some(codex_home_path.clone()),
+            ..Default::default()
+        })
+        .await?;
+    let resume_err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_error_message(RequestId::Integer(resume_id)),
+    )
+    .await??;
+    assert!(
+        resume_err.error.message.contains("path is a directory"),
+        "unexpected resume error: {}",
+        resume_err.error.message
+    );
+    assert!(
+        !resume_err.error.message.contains("Is a directory"),
+        "resume should reject the directory before rollout reading: {}",
+        resume_err.error.message
+    );
+
+    let fork_id = app_server
+        .send_thread_fork_request(ThreadForkParams {
+            thread_id: side_thread_id,
+            path: Some(codex_home_path),
+            ..Default::default()
+        })
+        .await?;
+    let fork_err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_error_message(RequestId::Integer(fork_id)),
+    )
+    .await??;
+    assert!(
+        fork_err.error.message.contains("path is a directory"),
+        "unexpected fork error: {}",
+        fork_err.error.message
+    );
+    assert!(
+        !fork_err.error.message.contains("Is a directory"),
+        "fork should reject the directory before rollout reading: {}",
+        fork_err.error.message
+    );
+
+    Ok(())
+}
+
 // Helper to create a config.toml pointing at the mock model server.
 fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
     let config_toml = codex_home.join("config.toml");

codex-rs/rollout/src/compression.rs

@@ -939,16 +939,12 @@ mod path {
 
     pub(super) async fn existing_rollout_path(path: &Path) -> Option<PathBuf> {
         let plain_path = plain_rollout_path(path);
-        if tokio::fs::try_exists(plain_path.as_path())
-            .await
-            .unwrap_or(false)
+        if matches!(tokio::fs::metadata(plain_path.as_path()).await, Ok(metadata) if metadata.is_file())
         {
             return Some(plain_path);
         }
         let compressed_path = compressed_rollout_path(plain_path.as_path());
-        if tokio::fs::try_exists(compressed_path.as_path())
-            .await
-            .unwrap_or(false)
+        if matches!(tokio::fs::metadata(compressed_path.as_path()).await, Ok(metadata) if metadata.is_file())
         {
             return Some(compressed_path);
         }

codex-rs/thread-store/src/local/read_thread.rs

@@ -143,6 +143,25 @@ async fn resolve_requested_rollout_path(
     } else {
         rollout_path
     };
+    match tokio::fs::metadata(path.as_path()).await {
+        Ok(metadata) if metadata.is_dir() => {
+            return Err(ThreadStoreError::InvalidRequest {
+                message: format!(
+                    "failed to resolve rollout path `{}`: path is a directory",
+                    path.display()
+                ),
+            });
+        }
+        Ok(metadata) if !metadata.is_file() => {
+            return Err(ThreadStoreError::InvalidRequest {
+                message: format!(
+                    "failed to resolve rollout path `{}`: path is not a file",
+                    path.display()
+                ),
+            });
+        }
+        _ => {}
+    }
     let Some(path) = codex_rollout::existing_rollout_path(path.as_path()).await else {
         return Err(ThreadStoreError::InvalidRequest {
             message: format!(