try to fix :AGENTS.md bazel hack

### Motivation
- Avoid placing PATH entries under the system temp directory by creating
the helper directory under `CODEX_HOME` instead of
`std::env::temp_dir()`.
- Fail fast on unsafe configuration by rejecting `CODEX_HOME` values
that live under the system temp root to prevent writable PATH entries.

### Testing
- Ran `just fmt`, which completed with a non-blocking
`imports_granularity` warning.
- Ran `just fix -p codex-arg0` (Clippy fixes) which completed
successfully.
- Ran `cargo test -p codex-arg0` and the test run completed
successfully.

.bazelrc

@@ -0,0 +1,45 @@
+common --repo_env=BAZEL_DO_NOT_DETECT_CPP_TOOLCHAIN=1
+common --repo_env=BAZEL_NO_APPLE_CPP_TOOLCHAIN=1
+
+common --disk_cache=~/.cache/bazel-disk-cache
+common --repo_contents_cache=~/.cache/bazel-repo-contents-cache
+common --repository_cache=~/.cache/bazel-repo-cache
+
+common --experimental_platform_in_output_dir
+
+common --enable_platform_specific_config
+# TODO(zbarsky): We need to untangle these libc constraints to get linux remote builds working.
+common:linux --host_platform=//:local
+common --@rules_cc//cc/toolchains/args/archiver_flags:use_libtool_on_macos=False
+common --@toolchains_llvm_bootstrapped//config:experimental_stub_libgcc_s
+
+# We need to use the sh toolchain on windows so we don't send host bash paths to the linux executor.
+common:windows --@rules_rust//rust/settings:experimental_use_sh_toolchain_for_bootstrap_process_wrapper
+
+# TODO(zbarsky): rules_rust doesn't implement this flag properly with remote exec...
+# common --@rules_rust//rust/settings:pipelined_compilation
+
+common --incompatible_strict_action_env
+# Not ideal, but We need to allow dotslash to be found
+common --test_env=PATH=/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin
+
+common --test_output=errors
+common --bes_results_url=https://app.buildbuddy.io/invocation/
+common --bes_backend=grpcs://remote.buildbuddy.io
+common --remote_cache=grpcs://remote.buildbuddy.io
+common --remote_download_toplevel
+common --nobuild_runfile_links
+common --remote_timeout=3600
+common --noexperimental_throttle_remote_action_building
+common --experimental_remote_execution_keepalive
+common --grpc_keepalive_time=30s
+
+# This limits both in-flight executions and concurrent downloads. Even with high number
+# of jobs execution will still be limited by CPU cores, so this just pays a bit of
+# memory in exchange for higher download concurrency.
+common --jobs=30
+
+common:remote --extra_execution_platforms=//:rbe
+common:remote --remote_executor=grpcs://remote.buildbuddy.io
+common:remote --jobs=800
+

.github/workflows/Dockerfile.bazel

@@ -0,0 +1,20 @@
+FROM ubuntu:24.04
+
+# TODO(mbolin): Published to docker.io/mbolin491/codex-bazel:latest for
+# initial debugging, but we should publish to a more proper location.
+#
+# docker buildx create --use
+# docker buildx build --platform linux/amd64 -f .github/workflows/Dockerfile.bazel -t mbolin491/codex-bazel:latest --push .
+
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+    curl git python3 ca-certificates && \
+    rm -rf /var/lib/apt/lists/*
+
+# Install dotslash.
+RUN curl -LSfs "https://github.com/facebook/dotslash/releases/download/v0.5.8/dotslash-ubuntu-22.04.$(uname -m).tar.gz" | tar fxz - -C /usr/local/bin
+
+# Ubuntu 24.04 ships with user 'ubuntu' already created with UID 1000.
+USER ubuntu
+
+WORKDIR /workspace

.github/workflows/bazel.yml

@@ -0,0 +1,110 @@
+name: Bazel (experimental)
+
+# Note this workflow was originally derived from:
+# https://github.com/cerisier/toolchains_llvm_bootstrapped/blob/main/.github/workflows/ci.yaml
+
+on:
+  pull_request: {}
+  push:
+    branches:
+      - main
+  workflow_dispatch:
+
+concurrency:
+  # Cancel previous actions from the same PR or branch except 'main' branch.
+  # See https://docs.github.com/en/actions/using-jobs/using-concurrency and https://docs.github.com/en/actions/learn-github-actions/contexts for more info.
+  group: concurrency-group::${{ github.workflow }}::${{ github.event.pull_request.number > 0 && format('pr-{0}', github.event.pull_request.number) || github.ref_name }}${{ github.ref_name == 'main' && format('::{0}', github.run_id) || ''}}
+  cancel-in-progress: ${{ github.ref_name != 'main' }}
+jobs:
+  test:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          # macOS
+          - os: macos-15-xlarge
+            target: aarch64-apple-darwin
+          - os: macos-15-xlarge
+            target: x86_64-apple-darwin
+
+          # Linux
+          - os: ubuntu-24.04-arm
+            target: aarch64-unknown-linux-gnu
+          - os: ubuntu-24.04
+            target: x86_64-unknown-linux-gnu
+          - os: ubuntu-24.04-arm
+            target: aarch64-unknown-linux-musl
+          - os: ubuntu-24.04
+            target: x86_64-unknown-linux-musl
+          # TODO: Enable Windows once we fix the toolchain issues there.
+          #- os: windows-latest
+          #  target: x86_64-pc-windows-gnullvm
+    runs-on: ${{ matrix.os }}
+
+    # Configure a human readable name for each job
+    name: Local Bazel build on ${{ matrix.os }} for ${{ matrix.target }}
+
+    steps:
+      - uses: actions/checkout@v6
+
+      # Some integration tests rely on DotSlash being installed.
+      # See https://github.com/openai/codex/pull/7617.
+      - name: Install DotSlash
+        uses: facebook/install-dotslash@v2
+
+      - name: Make DotSlash available in PATH (Unix)
+        if: runner.os != 'Windows'
+        run: cp "$(which dotslash)" /usr/local/bin
+
+      - name: Make DotSlash available in PATH (Windows)
+        if: runner.os == 'Windows'
+        shell: pwsh
+        run: Copy-Item (Get-Command dotslash).Source -Destination "$env:LOCALAPPDATA\Microsoft\WindowsApps\dotslash.exe"
+
+      # Install Bazel via Bazelisk
+      - name: Set up Bazel
+        uses: bazelbuild/setup-bazelisk@v3
+
+      # TODO(mbolin): Bring this back once we have caching working. Currently,
+      # we never seem to get a cache hit but we still end up paying the cost of
+      # uploading at the end of the build, which takes over a minute!
+      #
+      # Cache build and external artifacts so that the next ci build is incremental.
+      # Because github action caches cannot be updated after a build, we need to
+      # store the contents of each build in a unique cache key, then fall back to loading
+      # it on the next ci run. We use hashFiles(...) in the key and restore-keys- with
+      # the prefix to load the most recent cache for the branch on a cache miss. You
+      # should customize the contents of hashFiles to capture any bazel input sources,
+      # although this doesn't need to be perfect. If none of the input sources change
+      # then a cache hit will load an existing cache and bazel won't have to do any work.
+      # In the case of a cache miss, you want the fallback cache to contain most of the
+      # previously built artifacts to minimize build time. The more precise you are with
+      # hashFiles sources the less work bazel will have to do.
+      # - name: Mount bazel caches
+      #   uses: actions/cache@v4
+      #   with:
+      #     path: |
+      #       ~/.cache/bazel-repo-cache
+      #       ~/.cache/bazel-repo-contents-cache
+      #     key: bazel-cache-${{ matrix.os }}-${{ hashFiles('**/BUILD.bazel', '**/*.bzl', 'MODULE.bazel') }}
+      #     restore-keys: |
+      #       bazel-cache-${{ matrix.os }}
+
+      - name: Configure Bazel startup args (Windows)
+        if: runner.os == 'Windows'
+        shell: pwsh
+        run: |
+          # Use a very short path to reduce argv/path length issues.
+          "BAZEL_STARTUP_ARGS=--output_user_root=C:\" | Out-File -FilePath $env:GITHUB_ENV -Encoding utf8 -Append
+
+      - name: bazel test //...
+        env:
+          BUILDBUDDY_API_KEY: ${{ secrets.BUILDBUDDY_API_KEY }}
+        shell: bash
+        run: |
+          bazel $BAZEL_STARTUP_ARGS --bazelrc=.github/workflows/ci.bazelrc test //... \
+            --build_metadata=REPO_URL=https://github.com/openai/codex.git \
+            --build_metadata=COMMIT_SHA=$(git rev-parse HEAD) \
+            --build_metadata=ROLE=CI \
+            --build_metadata=VISIBILITY=PUBLIC \
+            "--remote_header=x-buildbuddy-api-key=$BUILDBUDDY_API_KEY"

.github/workflows/ci.bazelrc

@@ -0,0 +1,15 @@
+common --remote_download_minimal
+common --nobuild_runfile_links
+common --keep_going
+
+# Prefer to run the build actions entirely remotely so we can dial up the concurrency.
+# Currently remote builds only work on Mac hosts, until we untangle the libc constraints mess on linux.
+common:macos --config=remote
+common:macos --strategy=remote
+
+# We have platform-specific tests, so execute the tests locally using the strongest sandboxing available on each platform.
+common:macos --strategy=TestRunner=darwin-sandbox,local
+# Note: linux-sandbox is stronger, but not available in GHA.
+common:linux --strategy=TestRunner=processwrapper-sandbox,local
+common:windows --strategy=TestRunner=local
+

.gitignore

@@ -9,6 +9,7 @@ node_modules
 
 # build
 dist/
+bazel-*
 build/
 out/
 storybook-static/

AGENTS.md

@@ -77,11 +77,11 @@ If you don’t have the tool:
 - Prefer deep equals comparisons whenever possible. Perform `assert_eq!()` on entire objects, rather than individual fields.
 - Avoid mutating process environment in tests; prefer passing environment-derived flags or dependencies from above.
 
-### Spawning workspace binaries in tests (Cargo vs Buck2)
+### Spawning workspace binaries in tests (Cargo vs Bazel)
 
 - Prefer `codex_utils_cargo_bin::cargo_bin("...")` over `assert_cmd::Command::cargo_bin(...)` or `escargot` when tests need to spawn first-party binaries.
-  - Under Buck2, `CARGO_BIN_EXE_*` may be project-relative (e.g. `buck-out/...`), which breaks if a test changes its working directory. `codex_utils_cargo_bin::cargo_bin` resolves to an absolute path first.
-- When locating fixture files under Buck2, avoid `env!("CARGO_MANIFEST_DIR")` (Buck codegen sets it to `"."`). Prefer deriving paths from `codex_utils_cargo_bin::buck_project_root()` when needed.
+  - Under Bazel, binaries and resources may live under runfiles; use `codex_utils_cargo_bin::cargo_bin` to resolve absolute paths that remain stable after `chdir`.
+- When locating fixture files or test resources under Bazel, avoid `env!("CARGO_MANIFEST_DIR")`. Prefer `codex_utils_cargo_bin::find_resource!` so paths resolve correctly under both Cargo and Bazel runfiles.
 
 ### Integration tests (core)
 

BUILD.bazel

@@ -0,0 +1,29 @@
+# We mark the local platform as glibc-compatible so that rust can grab a toolchain for us.
+# TODO(zbarsky): Upstream a better libc constraint into rules_rust.
+# We only enable this on linux though for sanity, and because it breaks remote execution.
+platform(
+    name = "local",
+    constraint_values = [
+        "@toolchains_llvm_bootstrapped//constraints/libc:gnu.2.28",
+    ],
+    parents = [
+        "@platforms//host",
+    ],
+)
+
+platform(
+    name = "rbe",
+    constraint_values = [
+        "@platforms//cpu:x86_64",
+        "@platforms//os:linux",
+        "@bazel_tools//tools/cpp:clang",
+        "@toolchains_llvm_bootstrapped//constraints/libc:gnu.2.28",
+    ],
+    exec_properties = {
+        # Ubuntu-based image that includes git, python3, dotslash, and other
+        # tools that various integration tests need.
+        # Verify at https://hub.docker.com/layers/mbolin491/codex-bazel/latest/images/sha256:8c9ff94187ea7c08a31e9a81f5fe8046ea3972a6768983c955c4079fa30567fb
+        "container-image": "docker://docker.io/mbolin491/codex-bazel@sha256:8c9ff94187ea7c08a31e9a81f5fe8046ea3972a6768983c955c4079fa30567fb",
+        "OSFamily": "Linux",
+    },
+)

MODULE.bazel

@@ -0,0 +1,122 @@
+bazel_dep(name = "platforms", version = "1.0.0")
+bazel_dep(name = "toolchains_llvm_bootstrapped", version = "0.3.1")
+archive_override(
+    module_name = "toolchains_llvm_bootstrapped",
+    integrity = "sha256-9ks21bgEqbQWmwUIvqeLA64+Jk6o4ZVjC8KxjVa2Vw8=",
+    strip_prefix = "toolchains_llvm_bootstrapped-e3775e66a7b6d287c705ca0cd24497ef4a77c503",
+    urls = ["https://github.com/cerisier/toolchains_llvm_bootstrapped/archive/e3775e66a7b6d287c705ca0cd24497ef4a77c503/master.tar.gz"],
+    patch_strip = 1,
+    patches = [
+        "//patches:llvm_toolchain_archive_params.patch",
+    ],
+)
+
+osx = use_extension("@toolchains_llvm_bootstrapped//toolchain/extension:osx.bzl", "osx")
+osx.framework(name = "ApplicationServices")
+osx.framework(name = "AppKit")
+osx.framework(name = "ColorSync")
+osx.framework(name = "CoreFoundation")
+osx.framework(name = "CoreGraphics")
+osx.framework(name = "CoreServices")
+osx.framework(name = "CoreText")
+osx.framework(name = "CFNetwork")
+osx.framework(name = "Foundation")
+osx.framework(name = "ImageIO")
+osx.framework(name = "Kernel")
+osx.framework(name = "OSLog")
+osx.framework(name = "Security")
+osx.framework(name = "SystemConfiguration")
+
+register_toolchains(
+    "@toolchains_llvm_bootstrapped//toolchain:all",
+)
+
+bazel_dep(name = "rules_cc", version = "0.2.16")
+bazel_dep(name = "rules_platform", version = "0.1.0")
+bazel_dep(name = "rules_rust", version = "0.68.1")
+single_version_override(
+    module_name = "rules_rust",
+    patch_strip = 1,
+    patches = [
+        "//patches:rules_rust.patch",
+        "//patches:rules_rust_windows_gnu.patch",
+        "//patches:rules_rust_musl.patch",
+    ],
+)
+
+RUST_TRIPLES = [
+    "aarch64-unknown-linux-musl",
+    "aarch64-apple-darwin",
+    "aarch64-pc-windows-gnullvm",
+    "x86_64-unknown-linux-musl",
+    "x86_64-apple-darwin",
+    "x86_64-pc-windows-gnullvm",
+]
+
+rust = use_extension("@rules_rust//rust:extensions.bzl", "rust")
+rust.toolchain(
+    edition = "2024",
+    extra_target_triples = RUST_TRIPLES,
+    versions = ["1.90.0"],
+)
+use_repo(rust, "rust_toolchains")
+
+register_toolchains("@rust_toolchains//:all")
+
+bazel_dep(name = "rules_rs", version = "0.0.23")
+
+crate = use_extension("@rules_rs//rs:extensions.bzl", "crate")
+crate.from_cargo(
+    cargo_lock = "//codex-rs:Cargo.lock",
+    cargo_toml = "//codex-rs:Cargo.toml",
+    platform_triples = RUST_TRIPLES,
+)
+
+bazel_dep(name = "openssl", version = "3.5.4.bcr.0")
+
+crate.annotation(
+    build_script_data = [
+        "@openssl//:gen_dir",
+    ],
+    build_script_env = {
+        "OPENSSL_DIR": "$(execpath @openssl//:gen_dir)",
+        "OPENSSL_NO_VENDOR": "1",
+        "OPENSSL_STATIC": "1",
+    },
+    crate = "openssl-sys",
+    data = ["@openssl//:gen_dir"],
+)
+
+inject_repo(crate, "openssl")
+
+# Fix readme inclusions
+crate.annotation(
+    crate = "windows-link",
+    patch_args = ["-p1"],
+    patches = [
+        "//patches:windows-link.patch"
+    ],
+)
+
+WINDOWS_IMPORT_LIB = """
+load("@rules_cc//cc:defs.bzl", "cc_import")
+
+cc_import(
+    name = "windows_import_lib",
+    static_library = glob(["lib/*.a"])[0],
+)
+"""
+
+crate.annotation(
+    additive_build_file_content = WINDOWS_IMPORT_LIB,
+    crate = "windows_x86_64_gnullvm",
+    gen_build_script = "off",
+    deps = [":windows_import_lib"],
+)
+crate.annotation(
+    additive_build_file_content = WINDOWS_IMPORT_LIB,
+    crate = "windows_aarch64_gnullvm",
+    gen_build_script = "off",
+    deps = [":windows_import_lib"],
+)
+use_repo(crate, "crates")

codex-cli/bin/codex.js

@@ -95,7 +95,6 @@ function detectPackageManager() {
     return "bun";
   }
 
-
   if (
     __dirname.includes(".bun/install/global") ||
     __dirname.includes(".bun\\install\\global")

codex-rs/BUILD.bazel

@@ -0,0 +1 @@
+

codex-rs/Cargo.lock

@@ -819,6 +819,8 @@ version = "1.2.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "deec109607ca693028562ed836a5f1c4b8bd77755c4e132fc5ce11b0b6211ae7"
 dependencies = [
+ "jobserver",
+ "libc",
  "shlex",
 ]
 
@@ -1127,6 +1129,7 @@ dependencies = [
  "codex-common",
  "codex-core",
  "codex-git",
+ "codex-utils-cargo-bin",
  "serde",
  "serde_json",
  "tempfile",
@@ -1153,7 +1156,6 @@ dependencies = [
  "codex-execpolicy",
  "codex-login",
  "codex-mcp-server",
- "codex-process-hardening",
  "codex-protocol",
  "codex-responses-api-proxy",
  "codex-rmcp-client",
@@ -1163,7 +1165,6 @@ dependencies = [
  "codex-utils-absolute-path",
  "codex-utils-cargo-bin",
  "codex-windows-sandbox",
- "ctor 0.5.0",
  "libc",
  "owo-colors",
  "predicates",
@@ -1197,6 +1198,7 @@ dependencies = [
  "tracing",
  "tracing-opentelemetry",
  "tracing-subscriber",
+ "zstd",
 ]
 
 [[package]]
@@ -1298,7 +1300,6 @@ dependencies = [
  "dunce",
  "encoding_rs",
  "env-flags",
- "escargot",
  "eventsource-stream",
  "futures",
  "http 1.3.1",
@@ -1348,6 +1349,19 @@ dependencies = [
  "which",
  "wildmatch",
  "wiremock",
+ "zstd",
+]
+
+[[package]]
+name = "codex-debug-client"
+version = "0.0.0"
+dependencies = [
+ "anyhow",
+ "clap",
+ "codex-app-server-protocol",
+ "pretty_assertions",
+ "serde",
+ "serde_json",
 ]
 
 [[package]]
@@ -1605,10 +1619,12 @@ dependencies = [
  "opentelemetry-otlp",
  "opentelemetry-semantic-conventions",
  "opentelemetry_sdk",
+ "pretty_assertions",
  "reqwest",
  "serde",
  "serde_json",
  "strum_macros 0.27.2",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
  "tracing-opentelemetry",
@@ -1876,6 +1892,7 @@ name = "codex-utils-cargo-bin"
 version = "0.0.0"
 dependencies = [
  "assert_cmd",
+ "path-absolutize",
  "thiserror 2.0.17",
 ]
 
@@ -2790,17 +2807,6 @@ version = "3.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dea2df4cf52843e0452895c455a1a2cfbb842a1e7329671acf418fdc53ed4c59"
 
-[[package]]
-name = "escargot"
-version = "0.5.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11c3aea32bc97b500c9ca6a72b768a26e558264303d101d3409cf6d57a9ed0cf"
-dependencies = [
- "log",
- "serde",
- "serde_json",
-]
-
 [[package]]
 name = "event-listener"
 version = "5.4.0"
@@ -3924,6 +3930,16 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130"
 
+[[package]]
+name = "jobserver"
+version = "0.1.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33"
+dependencies = [
+ "getrandom 0.3.3",
+ "libc",
+]
+
 [[package]]
 name = "js-sys"
 version = "0.3.77"
@@ -8809,6 +8825,34 @@ dependencies = [
  "syn 2.0.104",
 ]
 
+[[package]]
+name = "zstd"
+version = "0.13.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
+dependencies = [
+ "zstd-safe",
+]
+
+[[package]]
+name = "zstd-safe"
+version = "7.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
+dependencies = [
+ "zstd-sys",
+]
+
+[[package]]
+name = "zstd-sys"
+version = "2.0.16+zstd.1.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748"
+dependencies = [
+ "cc",
+ "pkg-config",
+]
+
 [[package]]
 name = "zune-core"
 version = "0.4.12"

codex-rs/Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "app-server",
     "app-server-protocol",
     "app-server-test-client",
+    "debug-client",
     "apply-patch",
     "arg0",
     "feedback",
@@ -134,7 +135,6 @@ dunce = "1.0.4"
 encoding_rs = "0.8.35"
 env-flags = "0.1.1"
 env_logger = "0.11.5"
-escargot = "0.5"
 eventsource-stream = "0.2.3"
 futures = { version = "0.3", default-features = false }
 http = "1.3.1"
@@ -218,6 +218,7 @@ tracing-subscriber = "0.3.22"
 tracing-test = "0.2.5"
 tree-sitter = "0.25.10"
 tree-sitter-bash = "0.25"
+zstd = "0.13"
 tree-sitter-highlight = "0.25.10"
 ts-rs = "11"
 tui-scrollbar = "0.2.1"

codex-rs/ansi-escape/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "ansi-escape",
+    crate_name = "codex_ansi_escape",
+)

codex-rs/app-server-protocol/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "app-server-protocol",
+    crate_name = "codex_app_server_protocol",
+)

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -109,6 +109,10 @@ client_request_definitions! {
         params: v2::ThreadResumeParams,
         response: v2::ThreadResumeResponse,
     },
+    ThreadFork => "thread/fork" {
+        params: v2::ThreadForkParams,
+        response: v2::ThreadForkResponse,
+    },
     ThreadArchive => "thread/archive" {
         params: v2::ThreadArchiveParams,
         response: v2::ThreadArchiveResponse,
@@ -121,6 +125,10 @@ client_request_definitions! {
         params: v2::ThreadListParams,
         response: v2::ThreadListResponse,
     },
+    ThreadLoadedList => "thread/loaded/list" {
+        params: v2::ThreadLoadedListParams,
+        response: v2::ThreadLoadedListResponse,
+    },
     SkillsList => "skills/list" {
         params: v2::SkillsListParams,
         response: v2::SkillsListResponse,
@@ -197,6 +205,11 @@ client_request_definitions! {
         response: v2::ConfigWriteResponse,
     },
 
+    ConfigRequirementsRead => "configRequirements/read" {
+        params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
+        response: v2::ConfigRequirementsReadResponse,
+    },
+
     GetAccount => "account/read" {
         params: v2::GetAccountParams,
         response: v2::GetAccountResponse,
@@ -221,6 +234,11 @@ client_request_definitions! {
         params: v1::ResumeConversationParams,
         response: v1::ResumeConversationResponse,
     },
+    /// Fork a recorded Codex conversation into a new session.
+    ForkConversation {
+        params: v1::ForkConversationParams,
+        response: v1::ForkConversationResponse,
+    },
     ArchiveConversation {
         params: v1::ArchiveConversationParams,
         response: v1::ArchiveConversationResponse,
@@ -711,6 +729,22 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn serialize_config_requirements_read() -> Result<()> {
+        let request = ClientRequest::ConfigRequirementsRead {
+            request_id: RequestId::Integer(1),
+            params: None,
+        };
+        assert_eq!(
+            json!({
+                "method": "configRequirements/read",
+                "id": 1,
+            }),
+            serde_json::to_value(&request)?,
+        );
+        Ok(())
+    }
+
     #[test]
     fn serialize_account_login_api_key() -> Result<()> {
         let request = ClientRequest::LoginAccount {

codex-rs/app-server-protocol/src/protocol/v1.rs

@@ -83,6 +83,15 @@ pub struct ResumeConversationResponse {
     pub rollout_path: PathBuf,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+pub struct ForkConversationResponse {
+    pub conversation_id: ThreadId,
+    pub model: String,
+    pub initial_messages: Option<Vec<EventMsg>>,
+    pub rollout_path: PathBuf,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(untagged)]
 pub enum GetConversationSummaryParams {
@@ -148,6 +157,14 @@ pub struct ResumeConversationParams {
     pub overrides: Option<NewConversationParams>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+pub struct ForkConversationParams {
+    pub path: Option<PathBuf>,
+    pub conversation_id: Option<ThreadId>,
+    pub overrides: Option<NewConversationParams>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 pub struct AddConversationSubscriptionResponse {

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

@@ -453,6 +453,22 @@ pub struct ConfigReadResponse {
     pub layers: Option<Vec<ConfigLayer>>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ConfigRequirements {
+    pub allowed_approval_policies: Option<Vec<AskForApproval>>,
+    pub allowed_sandbox_modes: Option<Vec<SandboxMode>>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ConfigRequirementsReadResponse {
+    /// Null if no requirements are configured (e.g. no requirements.toml/MDM entries).
+    pub requirements: Option<ConfigRequirements>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -1064,6 +1080,47 @@ pub struct ThreadResumeResponse {
     pub reasoning_effort: Option<ReasoningEffort>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+/// There are two ways to fork a thread:
+/// 1. By thread_id: load the thread from disk by thread_id and fork it into a new thread.
+/// 2. By path: load the thread from disk by path and fork it into a new thread.
+///
+/// If using path, the thread_id param will be ignored.
+///
+/// Prefer using thread_id whenever possible.
+pub struct ThreadForkParams {
+    pub thread_id: String,
+
+    /// [UNSTABLE] Specify the rollout path to fork from.
+    /// If specified, the thread_id param will be ignored.
+    pub path: Option<PathBuf>,
+
+    /// Configuration overrides for the forked thread, if any.
+    pub model: Option<String>,
+    pub model_provider: Option<String>,
+    pub cwd: Option<String>,
+    pub approval_policy: Option<AskForApproval>,
+    pub sandbox: Option<SandboxMode>,
+    pub config: Option<HashMap<String, serde_json::Value>>,
+    pub base_instructions: Option<String>,
+    pub developer_instructions: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadForkResponse {
+    pub thread: Thread,
+    pub model: String,
+    pub model_provider: String,
+    pub cwd: PathBuf,
+    pub approval_policy: AskForApproval,
+    pub sandbox: SandboxPolicy,
+    pub reasoning_effort: Option<ReasoningEffort>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -1123,6 +1180,27 @@ pub struct ThreadListResponse {
     pub next_cursor: Option<String>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadLoadedListParams {
+    /// Opaque pagination cursor returned by a previous call.
+    pub cursor: Option<String>,
+    /// Optional page size; defaults to no limit.
+    pub limit: Option<u32>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadLoadedListResponse {
+    /// Thread ids for sessions currently loaded in memory.
+    pub data: Vec<String>,
+    /// Opaque cursor to pass to the next call to continue after the last item.
+    /// if None, there are no more items to return.
+    pub next_cursor: Option<String>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -1238,7 +1316,7 @@ pub struct Thread {
     pub source: SessionSource,
     /// Optional Git metadata captured when the thread was created.
     pub git_info: Option<GitInfo>,
-    /// Only populated on `thread/resume` and `thread/rollback` responses.
+    /// Only populated on `thread/resume`, `thread/rollback`, `thread/fork` responses.
     /// For all other responses and notifications returning a Thread,
     /// the turns field will be an empty list.
     pub turns: Vec<Turn>,
@@ -1314,7 +1392,7 @@ impl From<CoreTokenUsage> for TokenUsageBreakdown {
 #[ts(export_to = "v2/")]
 pub struct Turn {
     pub id: String,
-    /// Only populated on a `thread/resume` response.
+    /// Only populated on a `thread/resume` or `thread/fork` response.
     /// For all other responses and notifications returning a Turn,
     /// the items field will be an empty list.
     pub items: Vec<ThreadItem>,
@@ -1460,6 +1538,7 @@ pub enum UserInput {
     Text { text: String },
     Image { url: String },
     LocalImage { path: PathBuf },
+    Skill { name: String, path: PathBuf },
 }
 
 impl UserInput {
@@ -1468,6 +1547,7 @@ impl UserInput {
             UserInput::Text { text } => CoreUserInput::Text { text },
             UserInput::Image { url } => CoreUserInput::Image { image_url: url },
             UserInput::LocalImage { path } => CoreUserInput::LocalImage { path },
+            UserInput::Skill { name, path } => CoreUserInput::Skill { name, path },
         }
     }
 }
@@ -1478,6 +1558,7 @@ impl From<CoreUserInput> for UserInput {
             CoreUserInput::Text { text } => UserInput::Text { text },
             CoreUserInput::Image { image_url } => UserInput::Image { url: image_url },
             CoreUserInput::LocalImage { path } => UserInput::LocalImage { path },
+            CoreUserInput::Skill { name, path } => UserInput::Skill { name, path },
             _ => unreachable!("unsupported user input variant"),
         }
     }
@@ -2063,6 +2144,10 @@ mod tests {
                 CoreUserInput::LocalImage {
                     path: PathBuf::from("local/image.png"),
                 },
+                CoreUserInput::Skill {
+                    name: "skill-creator".to_string(),
+                    path: PathBuf::from("/repo/.codex/skills/skill-creator/SKILL.md"),
+                },
             ],
         });
 
@@ -2080,6 +2165,10 @@ mod tests {
                     UserInput::LocalImage {
                         path: PathBuf::from("local/image.png"),
                     },
+                    UserInput::Skill {
+                        name: "skill-creator".to_string(),
+                        path: PathBuf::from("/repo/.codex/skills/skill-creator/SKILL.md"),
+                    },
                 ],
             }
         );

codex-rs/app-server-test-client/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "codex-app-server-test-client",
+    crate_name = "codex_app_server_test_client",
+)

codex-rs/app-server-test-client/src/main.rs

@@ -545,7 +545,7 @@ impl CodexClient {
                         print!("{}", event.delta);
                         std::io::stdout().flush().ok();
                     }
-                    EventMsg::TaskComplete(event) => {
+                    EventMsg::TurnComplete(event) => {
                         println!("\n[task complete: {event:?}]");
                         break;
                     }

codex-rs/app-server/BUILD.bazel

@@ -0,0 +1,8 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "app-server",
+    crate_name = "codex_app_server",
+    integration_deps_extra = ["//codex-rs/app-server/tests/common:common"],
+    test_tags = ["no-sandbox"],
+)

codex-rs/app-server/README.md

@@ -41,7 +41,7 @@ Use the thread APIs to create, list, or archive conversations. Drive a conversat
 ## Lifecycle Overview
 
 - Initialize once: Immediately after launching the codex app-server process, send an `initialize` request with your client metadata, then emit an `initialized` notification. Any other request before this handshake gets rejected.
-- Start (or resume) a thread: Call `thread/start` to open a fresh conversation. The response returns the thread object and you’ll also get a `thread/started` notification. If you’re continuing an existing conversation, call `thread/resume` with its ID instead.
+- Start (or resume) a thread: Call `thread/start` to open a fresh conversation. The response returns the thread object and you’ll also get a `thread/started` notification. If you’re continuing an existing conversation, call `thread/resume` with its ID instead. If you want to branch from an existing conversation, call `thread/fork` to create a new thread id with copied history.
 - Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, etc. This immediately returns the new turn object and triggers a `turn/started` notification.
 - Stream events: After `turn/start`, keep reading JSON-RPC notifications on stdout. You’ll see `item/started`, `item/completed`, deltas like `item/agentMessage/delta`, tool progress, etc. These represent streaming model output plus any side effects (commands, tool calls, reasoning notes).
 - Finish the turn: When the model is done (or the turn is interrupted via making the `turn/interrupt` call), the server sends `turn/completed` with the final turn state and token usage.
@@ -52,6 +52,10 @@ Clients must send a single `initialize` request before invoking any other method
 
 Applications building on top of `codex app-server` should identify themselves via the `clientInfo` parameter.
 
+**Important**: `clientInfo.name` is used to identify the client for the OpenAI Compliance Logs Platform. If
+you are developing a new Codex integration that is intended for enterprise use, please contact us to get it
+added to a known clients list. For more context: https://chatgpt.com/admin/api-reference#tag/Logs:-Codex
+
 Example (from OpenAI's official VSCode extension):
 
 ```json
@@ -60,7 +64,7 @@ Example (from OpenAI's official VSCode extension):
   "id": 0,
   "params": {
     "clientInfo": {
-      "name": "codex-vscode",
+      "name": "codex_vscode",
       "title": "Codex VS Code Extension",
       "version": "0.1.0"
     }
@@ -72,7 +76,9 @@ Example (from OpenAI's official VSCode extension):
 
 - `thread/start` — create a new thread; emits `thread/started` 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; emits `thread/started` 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` filtering.
+- `thread/loaded/list` — list the thread ids currently loaded in memory.
 - `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success.
 - `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
 - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications.
@@ -88,6 +94,7 @@ Example (from OpenAI's official VSCode extension):
 - `config/read` — fetch the effective config on disk after resolving config layering.
 - `config/value/write` — write a single config key/value to the user's config.toml on disk.
 - `config/batchWrite` — apply multiple config edits atomically to the user's config.toml on disk.
+- `configRequirements/read` — fetch the loaded requirements allow-lists from `requirements.toml` and/or MDM (or `null` if none are configured).
 
 ### Example: Start or resume a thread
 
@@ -120,6 +127,14 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev
 { "id": 11, "result": { "thread": { "id": "thr_123", … } } }
 ```
 
+To branch from a stored session, call `thread/fork` with the `thread.id`. This creates a new thread id and emits a `thread/started` notification for it:
+
+```json
+{ "method": "thread/fork", "id": 12, "params": { "threadId": "thr_123" } }
+{ "id": 12, "result": { "thread": { "id": "thr_456", … } } }
+{ "method": "thread/started", "params": { "thread": { … } } }
+```
+
 ### Example: List threads (with pagination & filters)
 
 `thread/list` lets you render a history UI. Pass any combination of:
@@ -146,6 +161,17 @@ Example:
 
 When `nextCursor` is `null`, you’ve reached the final page.
 
+### Example: List loaded threads
+
+`thread/loaded/list` returns thread ids currently loaded in memory. This is useful when you want to check which sessions are active without scanning rollouts on disk.
+
+```json
+{ "method": "thread/loaded/list", "id": 21 }
+{ "id": 21, "result": {
+    "data": ["thr_123", "thr_456"]
+} }
+```
+
 ### Example: Archive a thread
 
 Use `thread/archive` to move the persisted rollout (stored as a JSONL file on disk) into the archived sessions directory.
@@ -200,13 +226,14 @@ You can optionally specify config overrides on the new turn. If specified, these
 
 ### Example: Start a turn (invoke a skill)
 
-Invoke a skill by sending a text input that begins with `$<skill-name>`.
+Invoke a skill explicitly by including `$<skill-name>` in the text input and adding a `skill` input item alongside it.
 
 ```json
 { "method": "turn/start", "id": 33, "params": {
     "threadId": "thr_123",
     "input": [
-        { "type": "text", "text": "$skill-creator Add a new skill for triaging flaky CI and include step-by-step usage." }
+        { "type": "text", "text": "$skill-creator Add a new skill for triaging flaky CI and include step-by-step usage." },
+        { "type": "skill", "name": "skill-creator", "path": "/Users/me/.codex/skills/skill-creator/SKILL.md" }
     ]
 } }
 { "id": 33, "result": { "turn": {
@@ -428,7 +455,23 @@ UI guidance for IDEs: surface an approval dialog as soon as the request arrives.
 
 ## Skills
 
-Skills are invoked by sending a text input that starts with `$<skill-name>`. The rest of the text is passed to the skill as its input.
+Invoke a skill by including `$<skill-name>` in the text input. Add a `skill` input item (recommended) so the backend injects full skill instructions instead of relying on the model to resolve the name.
+
+```json
+{
+  "method": "turn/start",
+  "id": 101,
+  "params": {
+    "threadId": "thread-1",
+    "input": [
+      { "type": "text", "text": "$skill-creator Add a new skill for triaging flaky CI." },
+      { "type": "skill", "name": "skill-creator", "path": "/Users/me/.codex/skills/skill-creator/SKILL.md" }
+    ]
+  }
+}
+```
+
+If you omit the `skill` item, the model will still parse the `$<skill-name>` marker and try to locate the skill, which can add latency.
 
 Example:
 

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

@@ -106,7 +106,7 @@ pub(crate) async fn apply_bespoke_event_handling(
         msg,
     } = event;
     match msg {
-        EventMsg::TaskComplete(_ev) => {
+        EventMsg::TurnComplete(_ev) => {
             handle_turn_complete(
                 conversation_id,
                 event_turn_id,

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

@@ -28,6 +28,8 @@ use codex_app_server_protocol::ConversationSummary;
 use codex_app_server_protocol::ExecOneOffCommandResponse;
 use codex_app_server_protocol::FeedbackUploadParams;
 use codex_app_server_protocol::FeedbackUploadResponse;
+use codex_app_server_protocol::ForkConversationParams;
+use codex_app_server_protocol::ForkConversationResponse;
 use codex_app_server_protocol::FuzzyFileSearchParams;
 use codex_app_server_protocol::FuzzyFileSearchResponse;
 use codex_app_server_protocol::GetAccountParams;
@@ -86,9 +88,13 @@ use codex_app_server_protocol::SkillsListResponse;
 use codex_app_server_protocol::Thread;
 use codex_app_server_protocol::ThreadArchiveParams;
 use codex_app_server_protocol::ThreadArchiveResponse;
+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;
+use codex_app_server_protocol::ThreadLoadedListParams;
+use codex_app_server_protocol::ThreadLoadedListResponse;
 use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadResumeResponse;
 use codex_app_server_protocol::ThreadRollbackParams;
@@ -124,6 +130,7 @@ use codex_core::config::ConfigService;
 use codex_core::config::edit::ConfigEditsBuilder;
 use codex_core::config::types::McpServerTransportConfig;
 use codex_core::default_client::get_codex_user_agent;
+use codex_core::error::CodexErr;
 use codex_core::exec::ExecParams;
 use codex_core::exec_env::create_env;
 use codex_core::features::Feature;
@@ -367,6 +374,9 @@ impl CodexMessageProcessor {
             ClientRequest::ThreadResume { request_id, params } => {
                 self.thread_resume(request_id, params).await;
             }
+            ClientRequest::ThreadFork { request_id, params } => {
+                self.thread_fork(request_id, params).await;
+            }
             ClientRequest::ThreadArchive { request_id, params } => {
                 self.thread_archive(request_id, params).await;
             }
@@ -376,6 +386,9 @@ impl CodexMessageProcessor {
             ClientRequest::ThreadList { request_id, params } => {
                 self.thread_list(request_id, params).await;
             }
+            ClientRequest::ThreadLoadedList { request_id, params } => {
+                self.thread_loaded_list(request_id, params).await;
+            }
             ClientRequest::SkillsList { request_id, params } => {
                 self.skills_list(request_id, params).await;
             }
@@ -433,6 +446,9 @@ impl CodexMessageProcessor {
             ClientRequest::ResumeConversation { request_id, params } => {
                 self.handle_resume_conversation(request_id, params).await;
             }
+            ClientRequest::ForkConversation { request_id, params } => {
+                self.handle_fork_conversation(request_id, params).await;
+            }
             ClientRequest::ArchiveConversation { request_id, params } => {
                 self.archive_conversation(request_id, params).await;
             }
@@ -510,6 +526,9 @@ impl CodexMessageProcessor {
             | ClientRequest::ConfigBatchWrite { .. } => {
                 warn!("Config request reached CodexMessageProcessor unexpectedly");
             }
+            ClientRequest::ConfigRequirementsRead { .. } => {
+                warn!("ConfigRequirementsRead request reached CodexMessageProcessor unexpectedly");
+            }
             ClientRequest::GetAccountRateLimits {
                 request_id,
                 params: _,
@@ -582,7 +601,7 @@ impl CodexMessageProcessor {
                     .await;
 
                 let payload = AuthStatusChangeNotification {
-                    auth_method: self.auth_manager.auth().map(|auth| auth.mode),
+                    auth_method: self.auth_manager.auth_cached().map(|auth| auth.mode),
                 };
                 self.outgoing
                     .send_server_notification(ServerNotification::AuthStatusChange(payload))
@@ -612,7 +631,7 @@ impl CodexMessageProcessor {
                     .await;
 
                 let payload_v2 = AccountUpdatedNotification {
-                    auth_mode: self.auth_manager.auth().map(|auth| auth.mode),
+                    auth_mode: self.auth_manager.auth_cached().map(|auth| auth.mode),
                 };
                 self.outgoing
                     .send_server_notification(ServerNotification::AccountUpdated(payload_v2))
@@ -704,7 +723,7 @@ impl CodexMessageProcessor {
                             auth_manager.reload();
 
                             // Notify clients with the actual current auth mode.
-                            let current_auth_method = auth_manager.auth().map(|a| a.mode);
+                            let current_auth_method = auth_manager.auth_cached().map(|a| a.mode);
                             let payload = AuthStatusChangeNotification {
                                 auth_method: current_auth_method,
                             };
@@ -794,7 +813,7 @@ impl CodexMessageProcessor {
                             auth_manager.reload();
 
                             // Notify clients with the actual current auth mode.
-                            let current_auth_method = auth_manager.auth().map(|a| a.mode);
+                            let current_auth_method = auth_manager.auth_cached().map(|a| a.mode);
                             let payload_v2 = AccountUpdatedNotification {
                                 auth_mode: current_auth_method,
                             };
@@ -906,7 +925,7 @@ impl CodexMessageProcessor {
         }
 
         // Reflect the current auth method after logout (likely None).
-        Ok(self.auth_manager.auth().map(|auth| auth.mode))
+        Ok(self.auth_manager.auth_cached().map(|auth| auth.mode))
     }
 
     async fn logout_v1(&mut self, request_id: RequestId) {
@@ -973,10 +992,10 @@ impl CodexMessageProcessor {
                 requires_openai_auth: Some(false),
             }
         } else {
-            match self.auth_manager.auth() {
+            match self.auth_manager.auth().await {
                 Some(auth) => {
                     let auth_mode = auth.mode;
-                    let (reported_auth_method, token_opt) = match auth.get_token().await {
+                    let (reported_auth_method, token_opt) = match auth.get_token() {
                         Ok(token) if !token.is_empty() => {
                             let tok = if include_token { Some(token) } else { None };
                             (Some(auth_mode), tok)
@@ -1021,7 +1040,7 @@ impl CodexMessageProcessor {
             return;
         }
 
-        let account = match self.auth_manager.auth() {
+        let account = match self.auth_manager.auth_cached() {
             Some(auth) => Some(match auth.mode {
                 AuthMode::ApiKey => Account::ApiKey {},
                 AuthMode::ChatGPT => {
@@ -1075,7 +1094,7 @@ impl CodexMessageProcessor {
     }
 
     async fn fetch_account_rate_limits(&self) -> Result<CoreRateLimitSnapshot, JSONRPCErrorError> {
-        let Some(auth) = self.auth_manager.auth() else {
+        let Some(auth) = self.auth_manager.auth().await else {
             return Err(JSONRPCErrorError {
                 code: INVALID_REQUEST_ERROR_CODE,
                 message: "codex account authentication required to read rate limits".to_string(),
@@ -1092,7 +1111,6 @@ impl CodexMessageProcessor {
         }
 
         let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
-            .await
             .map_err(|err| JSONRPCErrorError {
                 code: INTERNAL_ERROR_CODE,
                 message: format!("failed to construct backend client: {err}"),
@@ -1132,7 +1150,10 @@ impl CodexMessageProcessor {
 
     async fn get_user_info(&self, request_id: RequestId) {
         // Read alleged user email from cached auth (best-effort; not verified).
-        let alleged_user_email = self.auth_manager.auth().and_then(|a| a.get_account_email());
+        let alleged_user_email = self
+            .auth_manager
+            .auth_cached()
+            .and_then(|a| a.get_account_email());
 
         let response = UserInfoResponse { alleged_user_email };
         self.outgoing.send_response(request_id, response).await;
@@ -1588,6 +1609,61 @@ impl CodexMessageProcessor {
         self.outgoing.send_response(request_id, response).await;
     }
 
+    async fn thread_loaded_list(&self, request_id: RequestId, params: ThreadLoadedListParams) {
+        let ThreadLoadedListParams { cursor, limit } = params;
+        let mut data = self
+            .thread_manager
+            .list_thread_ids()
+            .await
+            .into_iter()
+            .map(|thread_id| thread_id.to_string())
+            .collect::<Vec<_>>();
+
+        if data.is_empty() {
+            let response = ThreadLoadedListResponse {
+                data,
+                next_cursor: None,
+            };
+            self.outgoing.send_response(request_id, response).await;
+            return;
+        }
+
+        data.sort();
+        let total = data.len();
+        let start = match cursor {
+            Some(cursor) => {
+                let cursor = match ThreadId::from_string(&cursor) {
+                    Ok(id) => id.to_string(),
+                    Err(_) => {
+                        let error = JSONRPCErrorError {
+                            code: INVALID_REQUEST_ERROR_CODE,
+                            message: format!("invalid cursor: {cursor}"),
+                            data: None,
+                        };
+                        self.outgoing.send_error(request_id, error).await;
+                        return;
+                    }
+                };
+                match data.binary_search(&cursor) {
+                    Ok(idx) => idx + 1,
+                    Err(idx) => idx,
+                }
+            }
+            None => 0,
+        };
+
+        let effective_limit = limit.unwrap_or(total as u32).max(1) as usize;
+        let end = start.saturating_add(effective_limit).min(total);
+        let page = data[start..end].to_vec();
+        let next_cursor = page.last().filter(|_| end < total).cloned();
+
+        let response = ThreadLoadedListResponse {
+            data: page,
+            next_cursor,
+        };
+        self.outgoing.send_response(request_id, response).await;
+    }
+
     async fn thread_resume(&mut self, request_id: RequestId, params: ThreadResumeParams) {
         let ThreadResumeParams {
             thread_id,
@@ -1793,6 +1869,198 @@ impl CodexMessageProcessor {
         }
     }
 
+    async fn thread_fork(&mut self, request_id: RequestId, params: ThreadForkParams) {
+        let ThreadForkParams {
+            thread_id,
+            path,
+            model,
+            model_provider,
+            cwd,
+            approval_policy,
+            sandbox,
+            config: cli_overrides,
+            base_instructions,
+            developer_instructions,
+        } = params;
+
+        let overrides_requested = model.is_some()
+            || model_provider.is_some()
+            || cwd.is_some()
+            || approval_policy.is_some()
+            || sandbox.is_some()
+            || cli_overrides.is_some()
+            || base_instructions.is_some()
+            || developer_instructions.is_some();
+
+        let config = if overrides_requested {
+            let overrides = self.build_thread_config_overrides(
+                model,
+                model_provider,
+                cwd,
+                approval_policy,
+                sandbox,
+                base_instructions,
+                developer_instructions,
+            );
+
+            // Persist windows sandbox feature.
+            let mut cli_overrides = cli_overrides.unwrap_or_default();
+            if cfg!(windows) && self.config.features.enabled(Feature::WindowsSandbox) {
+                cli_overrides.insert(
+                    "features.experimental_windows_sandbox".to_string(),
+                    serde_json::json!(true),
+                );
+            }
+
+            match derive_config_from_params(&self.cli_overrides, Some(cli_overrides), overrides)
+                .await
+            {
+                Ok(config) => config,
+                Err(err) => {
+                    let error = JSONRPCErrorError {
+                        code: INVALID_REQUEST_ERROR_CODE,
+                        message: format!("error deriving config: {err}"),
+                        data: None,
+                    };
+                    self.outgoing.send_error(request_id, error).await;
+                    return;
+                }
+            }
+        } else {
+            self.config.as_ref().clone()
+        };
+
+        let rollout_path = if let Some(path) = path {
+            path
+        } else {
+            let existing_thread_id = match ThreadId::from_string(&thread_id) {
+                Ok(id) => id,
+                Err(err) => {
+                    let error = JSONRPCErrorError {
+                        code: INVALID_REQUEST_ERROR_CODE,
+                        message: format!("invalid thread id: {err}"),
+                        data: None,
+                    };
+                    self.outgoing.send_error(request_id, error).await;
+                    return;
+                }
+            };
+
+            match find_thread_path_by_id_str(
+                &self.config.codex_home,
+                &existing_thread_id.to_string(),
+            )
+            .await
+            {
+                Ok(Some(p)) => p,
+                Ok(None) => {
+                    self.send_invalid_request_error(
+                        request_id,
+                        format!("no rollout found for thread id {existing_thread_id}"),
+                    )
+                    .await;
+                    return;
+                }
+                Err(err) => {
+                    self.send_invalid_request_error(
+                        request_id,
+                        format!("failed to locate thread id {existing_thread_id}: {err}"),
+                    )
+                    .await;
+                    return;
+                }
+            }
+        };
+
+        let fallback_model_provider = config.model_provider_id.clone();
+
+        let NewThread {
+            thread_id,
+            session_configured,
+            ..
+        } = match self
+            .thread_manager
+            .fork_thread(usize::MAX, config, rollout_path.clone())
+            .await
+        {
+            Ok(thread) => thread,
+            Err(err) => {
+                let (code, message) = match err {
+                    CodexErr::Io(_) | CodexErr::Json(_) => (
+                        INVALID_REQUEST_ERROR_CODE,
+                        format!("failed to load rollout `{}`: {err}", rollout_path.display()),
+                    ),
+                    CodexErr::InvalidRequest(message) => (INVALID_REQUEST_ERROR_CODE, message),
+                    _ => (INTERNAL_ERROR_CODE, format!("error forking thread: {err}")),
+                };
+                let error = JSONRPCErrorError {
+                    code,
+                    message,
+                    data: None,
+                };
+                self.outgoing.send_error(request_id, error).await;
+                return;
+            }
+        };
+
+        let SessionConfiguredEvent {
+            rollout_path,
+            initial_messages,
+            ..
+        } = session_configured;
+        // Auto-attach a conversation listener when forking a thread.
+        if let Err(err) = self
+            .attach_conversation_listener(thread_id, false, ApiVersion::V2)
+            .await
+        {
+            tracing::warn!(
+                "failed to attach listener for thread {}: {}",
+                thread_id,
+                err.message
+            );
+        }
+
+        let mut thread = match read_summary_from_rollout(
+            rollout_path.as_path(),
+            fallback_model_provider.as_str(),
+        )
+        .await
+        {
+            Ok(summary) => summary_to_thread(summary),
+            Err(err) => {
+                self.send_internal_error(
+                    request_id,
+                    format!(
+                        "failed to load rollout `{}` for thread {thread_id}: {err}",
+                        rollout_path.display()
+                    ),
+                )
+                .await;
+                return;
+            }
+        };
+        thread.turns = initial_messages
+            .as_deref()
+            .map_or_else(Vec::new, build_turns_from_event_msgs);
+
+        let response = ThreadForkResponse {
+            thread: thread.clone(),
+            model: session_configured.model,
+            model_provider: session_configured.model_provider_id,
+            cwd: session_configured.cwd,
+            approval_policy: session_configured.approval_policy.into(),
+            sandbox: session_configured.sandbox_policy.into(),
+            reasoning_effort: session_configured.reasoning_effort,
+        };
+
+        self.outgoing.send_response(request_id, response).await;
+
+        let notif = ThreadStartedNotification { thread };
+        self.outgoing
+            .send_server_notification(ServerNotification::ThreadStarted(notif))
+            .await;
+    }
+
     async fn get_thread_summary(
         &self,
         request_id: RequestId,
@@ -2416,6 +2684,166 @@ impl CodexMessageProcessor {
         }
     }
 
+    async fn handle_fork_conversation(
+        &self,
+        request_id: RequestId,
+        params: ForkConversationParams,
+    ) {
+        let ForkConversationParams {
+            path,
+            conversation_id,
+            overrides,
+        } = params;
+
+        // Derive a Config using the same logic as new conversation, honoring overrides if provided.
+        let config = match overrides {
+            Some(overrides) => {
+                let NewConversationParams {
+                    model,
+                    model_provider,
+                    profile,
+                    cwd,
+                    approval_policy,
+                    sandbox: sandbox_mode,
+                    config: cli_overrides,
+                    base_instructions,
+                    developer_instructions,
+                    compact_prompt,
+                    include_apply_patch_tool,
+                } = overrides;
+
+                // Persist windows sandbox feature.
+                let mut cli_overrides = cli_overrides.unwrap_or_default();
+                if cfg!(windows) && self.config.features.enabled(Feature::WindowsSandbox) {
+                    cli_overrides.insert(
+                        "features.experimental_windows_sandbox".to_string(),
+                        serde_json::json!(true),
+                    );
+                }
+
+                let overrides = ConfigOverrides {
+                    model,
+                    config_profile: profile,
+                    cwd: cwd.map(PathBuf::from),
+                    approval_policy,
+                    sandbox_mode,
+                    model_provider,
+                    codex_linux_sandbox_exe: self.codex_linux_sandbox_exe.clone(),
+                    base_instructions,
+                    developer_instructions,
+                    compact_prompt,
+                    include_apply_patch_tool,
+                    ..Default::default()
+                };
+
+                derive_config_from_params(&self.cli_overrides, Some(cli_overrides), overrides).await
+            }
+            None => Ok(self.config.as_ref().clone()),
+        };
+        let config = match config {
+            Ok(cfg) => cfg,
+            Err(err) => {
+                self.send_invalid_request_error(
+                    request_id,
+                    format!("error deriving config: {err}"),
+                )
+                .await;
+                return;
+            }
+        };
+
+        let rollout_path = if let Some(path) = path {
+            path
+        } else if let Some(conversation_id) = conversation_id {
+            match find_thread_path_by_id_str(&self.config.codex_home, &conversation_id.to_string())
+                .await
+            {
+                Ok(Some(found_path)) => found_path,
+                Ok(None) => {
+                    self.send_invalid_request_error(
+                        request_id,
+                        format!("no rollout found for conversation id {conversation_id}"),
+                    )
+                    .await;
+                    return;
+                }
+                Err(err) => {
+                    self.send_invalid_request_error(
+                        request_id,
+                        format!("failed to locate conversation id {conversation_id}: {err}"),
+                    )
+                    .await;
+                    return;
+                }
+            }
+        } else {
+            self.send_invalid_request_error(
+                request_id,
+                "either path or conversation id must be provided".to_string(),
+            )
+            .await;
+            return;
+        };
+
+        let NewThread {
+            thread_id,
+            session_configured,
+            ..
+        } = match self
+            .thread_manager
+            .fork_thread(usize::MAX, config, rollout_path.clone())
+            .await
+        {
+            Ok(thread) => thread,
+            Err(err) => {
+                let (code, message) = match err {
+                    CodexErr::Io(_) | CodexErr::Json(_) => (
+                        INVALID_REQUEST_ERROR_CODE,
+                        format!("failed to load rollout `{}`: {err}", rollout_path.display()),
+                    ),
+                    CodexErr::InvalidRequest(message) => (INVALID_REQUEST_ERROR_CODE, message),
+                    _ => (
+                        INTERNAL_ERROR_CODE,
+                        format!("error forking conversation: {err}"),
+                    ),
+                };
+                let error = JSONRPCErrorError {
+                    code,
+                    message,
+                    data: None,
+                };
+                self.outgoing.send_error(request_id, error).await;
+                return;
+            }
+        };
+
+        self.outgoing
+            .send_server_notification(ServerNotification::SessionConfigured(
+                SessionConfiguredNotification {
+                    session_id: session_configured.session_id,
+                    model: session_configured.model.clone(),
+                    reasoning_effort: session_configured.reasoning_effort,
+                    history_log_id: session_configured.history_log_id,
+                    history_entry_count: session_configured.history_entry_count,
+                    initial_messages: session_configured.initial_messages.clone(),
+                    rollout_path: session_configured.rollout_path.clone(),
+                },
+            ))
+            .await;
+        let initial_messages = session_configured
+            .initial_messages
+            .map(|msgs| msgs.into_iter().collect());
+
+        // Reply with conversation id + model and initial messages (when present)
+        let response = ForkConversationResponse {
+            conversation_id: thread_id,
+            model: session_configured.model.clone(),
+            initial_messages,
+            rollout_path: session_configured.rollout_path.clone(),
+        };
+        self.outgoing.send_response(request_id, response).await;
+    }
+
     async fn send_invalid_request_error(&self, request_id: RequestId, message: String) {
         let error = JSONRPCErrorError {
             code: INVALID_REQUEST_ERROR_CODE,
@@ -3145,7 +3573,11 @@ impl CodexMessageProcessor {
                         // JSON-serializing the `Event` as-is, but these should
                         // be migrated to be variants of `ServerNotification`
                         // instead.
-                        let method = format!("codex/event/{}", event.msg);
+                        let event_formatted = match &event.msg {
+                            EventMsg::TurnStarted(_) => "task_started",
+                            EventMsg::TurnComplete(_) => "task_complete",
+                            _ => &event.msg.to_string(),
+                        };
                         let mut params = match serde_json::to_value(event.clone()) {
                             Ok(serde_json::Value::Object(map)) => map,
                             Ok(_) => {
@@ -3164,7 +3596,7 @@ impl CodexMessageProcessor {
 
                         outgoing_for_task
                             .send_notification(OutgoingNotification {
-                                method,
+                                method: format!("codex/event/{event_formatted}"),
                                 params: Some(params.into()),
                             })
                             .await;
@@ -3250,6 +3682,16 @@ impl CodexMessageProcessor {
     }
 
     async fn upload_feedback(&self, request_id: RequestId, params: FeedbackUploadParams) {
+        if !self.config.feedback_enabled {
+            let error = JSONRPCErrorError {
+                code: INVALID_REQUEST_ERROR_CODE,
+                message: "sending feedback is disabled by configuration".to_string(),
+                data: None,
+            };
+            self.outgoing.send_error(request_id, error).await;
+            return;
+        }
+
         let FeedbackUploadParams {
             classification,
             reason,

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

@@ -3,13 +3,18 @@ use crate::error_code::INVALID_REQUEST_ERROR_CODE;
 use codex_app_server_protocol::ConfigBatchWriteParams;
 use codex_app_server_protocol::ConfigReadParams;
 use codex_app_server_protocol::ConfigReadResponse;
+use codex_app_server_protocol::ConfigRequirements;
+use codex_app_server_protocol::ConfigRequirementsReadResponse;
 use codex_app_server_protocol::ConfigValueWriteParams;
 use codex_app_server_protocol::ConfigWriteErrorCode;
 use codex_app_server_protocol::ConfigWriteResponse;
 use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::SandboxMode;
 use codex_core::config::ConfigService;
 use codex_core::config::ConfigServiceError;
+use codex_core::config_loader::ConfigRequirementsToml;
 use codex_core::config_loader::LoaderOverrides;
+use codex_core::config_loader::SandboxModeRequirement as CoreSandboxModeRequirement;
 use serde_json::json;
 use std::path::PathBuf;
 use toml::Value as TomlValue;
@@ -37,6 +42,19 @@ impl ConfigApi {
         self.service.read(params).await.map_err(map_error)
     }
 
+    pub(crate) async fn config_requirements_read(
+        &self,
+    ) -> Result<ConfigRequirementsReadResponse, JSONRPCErrorError> {
+        let requirements = self
+            .service
+            .read_requirements()
+            .await
+            .map_err(map_error)?
+            .map(map_requirements_toml_to_api);
+
+        Ok(ConfigRequirementsReadResponse { requirements })
+    }
+
     pub(crate) async fn write_value(
         &self,
         params: ConfigValueWriteParams,
@@ -52,6 +70,32 @@ impl ConfigApi {
     }
 }
 
+fn map_requirements_toml_to_api(requirements: ConfigRequirementsToml) -> ConfigRequirements {
+    ConfigRequirements {
+        allowed_approval_policies: requirements.allowed_approval_policies.map(|policies| {
+            policies
+                .into_iter()
+                .map(codex_app_server_protocol::AskForApproval::from)
+                .collect()
+        }),
+        allowed_sandbox_modes: requirements.allowed_sandbox_modes.map(|modes| {
+            modes
+                .into_iter()
+                .filter_map(map_sandbox_mode_requirement_to_api)
+                .collect()
+        }),
+    }
+}
+
+fn map_sandbox_mode_requirement_to_api(mode: CoreSandboxModeRequirement) -> Option<SandboxMode> {
+    match mode {
+        CoreSandboxModeRequirement::ReadOnly => Some(SandboxMode::ReadOnly),
+        CoreSandboxModeRequirement::WorkspaceWrite => Some(SandboxMode::WorkspaceWrite),
+        CoreSandboxModeRequirement::DangerFullAccess => Some(SandboxMode::DangerFullAccess),
+        CoreSandboxModeRequirement::ExternalSandbox => None,
+    }
+}
+
 fn map_error(err: ConfigServiceError) -> JSONRPCErrorError {
     if let Some(code) = err.write_error_code() {
         return config_write_error(code, err.to_string());
@@ -73,3 +117,38 @@ fn config_write_error(code: ConfigWriteErrorCode, message: impl Into<String>) ->
         })),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use codex_protocol::protocol::AskForApproval as CoreAskForApproval;
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn map_requirements_toml_to_api_converts_core_enums() {
+        let requirements = ConfigRequirementsToml {
+            allowed_approval_policies: Some(vec![
+                CoreAskForApproval::Never,
+                CoreAskForApproval::OnRequest,
+            ]),
+            allowed_sandbox_modes: Some(vec![
+                CoreSandboxModeRequirement::ReadOnly,
+                CoreSandboxModeRequirement::ExternalSandbox,
+            ]),
+        };
+
+        let mapped = map_requirements_toml_to_api(requirements);
+
+        assert_eq!(
+            mapped.allowed_approval_policies,
+            Some(vec![
+                codex_app_server_protocol::AskForApproval::Never,
+                codex_app_server_protocol::AskForApproval::OnRequest,
+            ])
+        );
+        assert_eq!(
+            mapped.allowed_sandbox_modes,
+            Some(vec![SandboxMode::ReadOnly]),
+        );
+    }
+}

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

@@ -92,13 +92,18 @@ pub async fn run_main(
 
     let feedback = CodexFeedback::new();
 
-    let otel =
-        codex_core::otel_init::build_provider(&config, env!("CARGO_PKG_VERSION")).map_err(|e| {
-            std::io::Error::new(
-                ErrorKind::InvalidData,
-                format!("error loading otel config: {e}"),
-            )
-        })?;
+    let otel = codex_core::otel_init::build_provider(
+        &config,
+        env!("CARGO_PKG_VERSION"),
+        Some("codex_app_server"),
+        false,
+    )
+    .map_err(|e| {
+        std::io::Error::new(
+            ErrorKind::InvalidData,
+            format!("error loading otel config: {e}"),
+        )
+    })?;
 
     // Install a simple subscriber so `tracing` output is visible.  Users can
     // control the log level with `RUST_LOG`.

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

@@ -21,8 +21,10 @@ use codex_core::AuthManager;
 use codex_core::ThreadManager;
 use codex_core::config::Config;
 use codex_core::config_loader::LoaderOverrides;
+use codex_core::default_client::SetOriginatorError;
 use codex_core::default_client::USER_AGENT_SUFFIX;
 use codex_core::default_client::get_codex_user_agent;
+use codex_core::default_client::set_default_originator;
 use codex_feedback::CodexFeedback;
 use codex_protocol::protocol::SessionSource;
 use toml::Value as TomlValue;
@@ -121,6 +123,27 @@ impl MessageProcessor {
                         title: _title,
                         version,
                     } = params.client_info;
+                    if let Err(error) = set_default_originator(name.clone()) {
+                        match error {
+                            SetOriginatorError::InvalidHeaderValue => {
+                                let error = JSONRPCErrorError {
+                                    code: INVALID_REQUEST_ERROR_CODE,
+                                    message: format!(
+                                        "Invalid clientInfo.name: '{name}'. Must be a valid HTTP header value."
+                                    ),
+                                    data: None,
+                                };
+                                self.outgoing.send_error(request_id, error).await;
+                                return;
+                            }
+                            SetOriginatorError::AlreadyInitialized => {
+                                // No-op. This is expected to happen if the originator is already set via env var.
+                                // TODO(owen): Once we remove support for CODEX_INTERNAL_ORIGINATOR_OVERRIDE,
+                                // this will be an unexpected state and we can return a JSON-RPC error indicating
+                                // internal server error.
+                            }
+                        }
+                    }
                     let user_agent_suffix = format!("{name}; {version}");
                     if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
                         *suffix = Some(user_agent_suffix);
@@ -158,6 +181,12 @@ impl MessageProcessor {
             ClientRequest::ConfigBatchWrite { request_id, params } => {
                 self.handle_config_batch_write(request_id, params).await;
             }
+            ClientRequest::ConfigRequirementsRead {
+                request_id,
+                params: _,
+            } => {
+                self.handle_config_requirements_read(request_id).await;
+            }
             other => {
                 self.codex_message_processor.process_request(other).await;
             }
@@ -210,4 +239,11 @@ impl MessageProcessor {
             Err(error) => self.outgoing.send_error(request_id, error).await,
         }
     }
+
+    async fn handle_config_requirements_read(&self, request_id: RequestId) {
+        match self.config_api.config_requirements_read().await {
+            Ok(response) => self.outgoing.send_response(request_id, response).await,
+            Err(error) => self.outgoing.send_error(request_id, error).await,
+        }
+    }
 }

codex-rs/app-server/tests/common/BUILD.bazel

@@ -0,0 +1,7 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "common",
+    crate_name = "app_test_support",
+    crate_srcs = glob(["*.rs"]),
+)

codex-rs/app-server/tests/common/lib.rs

@@ -17,9 +17,11 @@ pub use core_test_support::format_with_current_shell_non_login;
 pub use core_test_support::test_path_buf_with_windows;
 pub use core_test_support::test_tmp_path;
 pub use core_test_support::test_tmp_path_buf;
+pub use mcp_process::DEFAULT_CLIENT_NAME;
 pub use mcp_process::McpProcess;
-pub use mock_model_server::create_mock_chat_completions_server;
-pub use mock_model_server::create_mock_chat_completions_server_unchecked;
+pub use mock_model_server::create_mock_responses_server_repeating_assistant;
+pub use mock_model_server::create_mock_responses_server_sequence;
+pub use mock_model_server::create_mock_responses_server_sequence_unchecked;
 pub use models_cache::write_models_cache;
 pub use models_cache::write_models_cache_with_models;
 pub use responses::create_apply_patch_sse_response;

codex-rs/app-server/tests/common/mcp_process.rs

@@ -21,6 +21,7 @@ use codex_app_server_protocol::ConfigBatchWriteParams;
 use codex_app_server_protocol::ConfigReadParams;
 use codex_app_server_protocol::ConfigValueWriteParams;
 use codex_app_server_protocol::FeedbackUploadParams;
+use codex_app_server_protocol::ForkConversationParams;
 use codex_app_server_protocol::GetAccountParams;
 use codex_app_server_protocol::GetAuthStatusParams;
 use codex_app_server_protocol::InitializeParams;
@@ -43,7 +44,9 @@ use codex_app_server_protocol::SendUserTurnParams;
 use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::SetDefaultModelParams;
 use codex_app_server_protocol::ThreadArchiveParams;
+use codex_app_server_protocol::ThreadForkParams;
 use codex_app_server_protocol::ThreadListParams;
+use codex_app_server_protocol::ThreadLoadedListParams;
 use codex_app_server_protocol::ThreadResumeParams;
 use codex_app_server_protocol::ThreadRollbackParams;
 use codex_app_server_protocol::ThreadStartParams;
@@ -60,9 +63,11 @@ pub struct McpProcess {
     process: Child,
     stdin: ChildStdin,
     stdout: BufReader<ChildStdout>,
-    pending_user_messages: VecDeque<JSONRPCNotification>,
+    pending_messages: VecDeque<JSONRPCMessage>,
 }
 
+pub const DEFAULT_CLIENT_NAME: &str = "codex-app-server-tests";
+
 impl McpProcess {
     pub async fn new(codex_home: &Path) -> anyhow::Result<Self> {
         Self::new_with_env(codex_home, &[]).await
@@ -127,39 +132,68 @@ impl McpProcess {
             process,
             stdin,
             stdout,
-            pending_user_messages: VecDeque::new(),
+            pending_messages: VecDeque::new(),
         })
     }
 
     /// Performs the initialization handshake with the MCP server.
     pub async fn initialize(&mut self) -> anyhow::Result<()> {
-        let params = Some(serde_json::to_value(InitializeParams {
-            client_info: ClientInfo {
-                name: "codex-app-server-tests".to_string(),
+        let initialized = self
+            .initialize_with_client_info(ClientInfo {
+                name: DEFAULT_CLIENT_NAME.to_string(),
                 title: None,
                 version: "0.1.0".to_string(),
-            },
-        })?);
-        let req_id = self.send_request("initialize", params).await?;
-        let initialized = self.read_jsonrpc_message().await?;
-        let JSONRPCMessage::Response(response) = initialized else {
+            })
+            .await?;
+        let JSONRPCMessage::Response(_) = initialized else {
             unreachable!("expected JSONRPCMessage::Response for initialize, got {initialized:?}");
         };
-        if response.id != RequestId::Integer(req_id) {
-            anyhow::bail!(
-                "initialize response id mismatch: expected {}, got {:?}",
-                req_id,
-                response.id
-            );
-        }
-
-        // Send notifications/initialized to ack the response.
-        self.send_notification(ClientNotification::Initialized)
-            .await?;
-
         Ok(())
     }
 
+    /// Sends initialize with the provided client info and returns the response/error message.
+    pub async fn initialize_with_client_info(
+        &mut self,
+        client_info: ClientInfo,
+    ) -> anyhow::Result<JSONRPCMessage> {
+        let params = Some(serde_json::to_value(InitializeParams { client_info })?);
+        let request_id = self.send_request("initialize", params).await?;
+        let message = self.read_jsonrpc_message().await?;
+        match message {
+            JSONRPCMessage::Response(response) => {
+                if response.id != RequestId::Integer(request_id) {
+                    anyhow::bail!(
+                        "initialize response id mismatch: expected {}, got {:?}",
+                        request_id,
+                        response.id
+                    );
+                }
+
+                // Send notifications/initialized to ack the response.
+                self.send_notification(ClientNotification::Initialized)
+                    .await?;
+
+                Ok(JSONRPCMessage::Response(response))
+            }
+            JSONRPCMessage::Error(error) => {
+                if error.id != RequestId::Integer(request_id) {
+                    anyhow::bail!(
+                        "initialize error id mismatch: expected {}, got {:?}",
+                        request_id,
+                        error.id
+                    );
+                }
+                Ok(JSONRPCMessage::Error(error))
+            }
+            JSONRPCMessage::Notification(notification) => {
+                anyhow::bail!("unexpected JSONRPCMessage::Notification: {notification:?}");
+            }
+            JSONRPCMessage::Request(request) => {
+                anyhow::bail!("unexpected JSONRPCMessage::Request: {request:?}");
+            }
+        }
+    }
+
     /// Send a `newConversation` JSON-RPC request.
     pub async fn send_new_conversation_request(
         &mut self,
@@ -308,6 +342,15 @@ impl McpProcess {
         self.send_request("thread/resume", params).await
     }
 
+    /// Send a `thread/fork` JSON-RPC request.
+    pub async fn send_thread_fork_request(
+        &mut self,
+        params: ThreadForkParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("thread/fork", params).await
+    }
+
     /// Send a `thread/archive` JSON-RPC request.
     pub async fn send_thread_archive_request(
         &mut self,
@@ -335,6 +378,15 @@ impl McpProcess {
         self.send_request("thread/list", params).await
     }
 
+    /// Send a `thread/loaded/list` JSON-RPC request.
+    pub async fn send_thread_loaded_list_request(
+        &mut self,
+        params: ThreadLoadedListParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("thread/loaded/list", params).await
+    }
+
     /// Send a `model/list` JSON-RPC request.
     pub async fn send_list_models_request(
         &mut self,
@@ -353,6 +405,15 @@ impl McpProcess {
         self.send_request("resumeConversation", params).await
     }
 
+    /// Send a `forkConversation` JSON-RPC request.
+    pub async fn send_fork_conversation_request(
+        &mut self,
+        params: ForkConversationParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("forkConversation", params).await
+    }
+
     /// Send a `loginApiKey` JSON-RPC request.
     pub async fn send_login_api_key_request(
         &mut self,
@@ -544,27 +605,16 @@ impl McpProcess {
     pub async fn read_stream_until_request_message(&mut self) -> anyhow::Result<ServerRequest> {
         eprintln!("in read_stream_until_request_message()");
 
-        loop {
-            let message = self.read_jsonrpc_message().await?;
+        let message = self
+            .read_stream_until_message(|message| matches!(message, JSONRPCMessage::Request(_)))
+            .await?;
 
-            match message {
-                JSONRPCMessage::Notification(notification) => {
-                    eprintln!("notification: {notification:?}");
-                    self.enqueue_user_message(notification);
-                }
-                JSONRPCMessage::Request(jsonrpc_request) => {
-                    return jsonrpc_request.try_into().with_context(
-                        || "failed to deserialize ServerRequest from JSONRPCRequest",
-                    );
-                }
-                JSONRPCMessage::Error(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
-                }
-                JSONRPCMessage::Response(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Response: {message:?}");
-                }
-            }
-        }
+        let JSONRPCMessage::Request(jsonrpc_request) = message else {
+            unreachable!("expected JSONRPCMessage::Request, got {message:?}");
+        };
+        jsonrpc_request
+            .try_into()
+            .with_context(|| "failed to deserialize ServerRequest from JSONRPCRequest")
     }
 
     pub async fn read_stream_until_response_message(
@@ -573,52 +623,32 @@ impl McpProcess {
     ) -> anyhow::Result<JSONRPCResponse> {
         eprintln!("in read_stream_until_response_message({request_id:?})");
 
-        loop {
-            let message = self.read_jsonrpc_message().await?;
-            match message {
-                JSONRPCMessage::Notification(notification) => {
-                    eprintln!("notification: {notification:?}");
-                    self.enqueue_user_message(notification);
-                }
-                JSONRPCMessage::Request(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
-                }
-                JSONRPCMessage::Error(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
-                }
-                JSONRPCMessage::Response(jsonrpc_response) => {
-                    if jsonrpc_response.id == request_id {
-                        return Ok(jsonrpc_response);
-                    }
-                }
-            }
-        }
+        let message = self
+            .read_stream_until_message(|message| {
+                Self::message_request_id(message) == Some(&request_id)
+            })
+            .await?;
+
+        let JSONRPCMessage::Response(response) = message else {
+            unreachable!("expected JSONRPCMessage::Response, got {message:?}");
+        };
+        Ok(response)
     }
 
     pub async fn read_stream_until_error_message(
         &mut self,
         request_id: RequestId,
     ) -> anyhow::Result<JSONRPCError> {
-        loop {
-            let message = self.read_jsonrpc_message().await?;
-            match message {
-                JSONRPCMessage::Notification(notification) => {
-                    eprintln!("notification: {notification:?}");
-                    self.enqueue_user_message(notification);
-                }
-                JSONRPCMessage::Request(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
-                }
-                JSONRPCMessage::Response(_) => {
-                    // Keep scanning; we're waiting for an error with matching id.
-                }
-                JSONRPCMessage::Error(err) => {
-                    if err.id == request_id {
-                        return Ok(err);
-                    }
-                }
-            }
-        }
+        let message = self
+            .read_stream_until_message(|message| {
+                Self::message_request_id(message) == Some(&request_id)
+            })
+            .await?;
+
+        let JSONRPCMessage::Error(err) = message else {
+            unreachable!("expected JSONRPCMessage::Error, got {message:?}");
+        };
+        Ok(err)
     }
 
     pub async fn read_stream_until_notification_message(
@@ -627,46 +657,64 @@ impl McpProcess {
     ) -> anyhow::Result<JSONRPCNotification> {
         eprintln!("in read_stream_until_notification_message({method})");
 
-        if let Some(notification) = self.take_pending_notification_by_method(method) {
-            return Ok(notification);
+        let message = self
+            .read_stream_until_message(|message| {
+                matches!(
+                    message,
+                    JSONRPCMessage::Notification(notification) if notification.method == method
+                )
+            })
+            .await?;
+
+        let JSONRPCMessage::Notification(notification) = message else {
+            unreachable!("expected JSONRPCMessage::Notification, got {message:?}");
+        };
+        Ok(notification)
+    }
+
+    /// Clears any buffered messages so future reads only consider new stream items.
+    ///
+    /// We call this when e.g. we want to validate against the next turn and no longer care about
+    /// messages buffered from the prior turn.
+    pub fn clear_message_buffer(&mut self) {
+        self.pending_messages.clear();
+    }
+
+    /// Reads the stream until a message matches `predicate`, buffering any non-matching messages
+    /// for later reads.
+    async fn read_stream_until_message<F>(&mut self, predicate: F) -> anyhow::Result<JSONRPCMessage>
+    where
+        F: Fn(&JSONRPCMessage) -> bool,
+    {
+        if let Some(message) = self.take_pending_message(&predicate) {
+            return Ok(message);
         }
 
         loop {
             let message = self.read_jsonrpc_message().await?;
-            match message {
-                JSONRPCMessage::Notification(notification) => {
-                    if notification.method == method {
-                        return Ok(notification);
-                    }
-                    self.enqueue_user_message(notification);
-                }
-                JSONRPCMessage::Request(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
-                }
-                JSONRPCMessage::Error(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
-                }
-                JSONRPCMessage::Response(_) => {
-                    anyhow::bail!("unexpected JSONRPCMessage::Response: {message:?}");
-                }
+            if predicate(&message) {
+                return Ok(message);
             }
+            self.pending_messages.push_back(message);
         }
     }
 
-    fn take_pending_notification_by_method(&mut self, method: &str) -> Option<JSONRPCNotification> {
-        if let Some(pos) = self
-            .pending_user_messages
-            .iter()
-            .position(|notification| notification.method == method)
-        {
-            return self.pending_user_messages.remove(pos);
+    fn take_pending_message<F>(&mut self, predicate: &F) -> Option<JSONRPCMessage>
+    where
+        F: Fn(&JSONRPCMessage) -> bool,
+    {
+        if let Some(pos) = self.pending_messages.iter().position(predicate) {
+            return self.pending_messages.remove(pos);
         }
         None
     }
 
-    fn enqueue_user_message(&mut self, notification: JSONRPCNotification) {
-        if notification.method == "codex/event/user_message" {
-            self.pending_user_messages.push_back(notification);
+    fn message_request_id(message: &JSONRPCMessage) -> Option<&RequestId> {
+        match message {
+            JSONRPCMessage::Request(request) => Some(&request.id),
+            JSONRPCMessage::Response(response) => Some(&response.id),
+            JSONRPCMessage::Error(err) => Some(&err.id),
+            JSONRPCMessage::Notification(_) => None,
         }
     }
 }

codex-rs/app-server/tests/common/mock_model_server.rs

@@ -1,17 +1,18 @@
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 
+use core_test_support::responses;
 use wiremock::Mock;
 use wiremock::MockServer;
 use wiremock::Respond;
 use wiremock::ResponseTemplate;
 use wiremock::matchers::method;
-use wiremock::matchers::path;
+use wiremock::matchers::path_regex;
 
 /// Create a mock server that will provide the responses, in order, for
-/// requests to the `/v1/chat/completions` endpoint.
-pub async fn create_mock_chat_completions_server(responses: Vec<String>) -> MockServer {
-    let server = MockServer::start().await;
+/// requests to the `/v1/responses` endpoint.
+pub async fn create_mock_responses_server_sequence(responses: Vec<String>) -> MockServer {
+    let server = responses::start_mock_server().await;
 
     let num_calls = responses.len();
     let seq_responder = SeqResponder {
@@ -20,7 +21,7 @@ pub async fn create_mock_chat_completions_server(responses: Vec<String>) -> Mock
     };
 
     Mock::given(method("POST"))
-        .and(path("/v1/chat/completions"))
+        .and(path_regex(".*/responses$"))
         .respond_with(seq_responder)
         .expect(num_calls as u64)
         .mount(&server)
@@ -29,10 +30,10 @@ pub async fn create_mock_chat_completions_server(responses: Vec<String>) -> Mock
     server
 }
 
-/// Same as `create_mock_chat_completions_server` but does not enforce an
+/// Same as `create_mock_responses_server_sequence` but does not enforce an
 /// expectation on the number of calls.
-pub async fn create_mock_chat_completions_server_unchecked(responses: Vec<String>) -> MockServer {
-    let server = MockServer::start().await;
+pub async fn create_mock_responses_server_sequence_unchecked(responses: Vec<String>) -> MockServer {
+    let server = responses::start_mock_server().await;
 
     let seq_responder = SeqResponder {
         num_calls: AtomicUsize::new(0),
@@ -40,7 +41,7 @@ pub async fn create_mock_chat_completions_server_unchecked(responses: Vec<String
     };
 
     Mock::given(method("POST"))
-        .and(path("/v1/chat/completions"))
+        .and(path_regex(".*/responses$"))
         .respond_with(seq_responder)
         .mount(&server)
         .await;
@@ -57,10 +58,24 @@ impl Respond for SeqResponder {
     fn respond(&self, _: &wiremock::Request) -> ResponseTemplate {
         let call_num = self.num_calls.fetch_add(1, Ordering::SeqCst);
         match self.responses.get(call_num) {
-            Some(response) => ResponseTemplate::new(200)
-                .insert_header("content-type", "text/event-stream")
-                .set_body_raw(response.clone(), "text/event-stream"),
+            Some(response) => responses::sse_response(response.clone()),
             None => panic!("no response for {call_num}"),
         }
     }
 }
+
+/// Create a mock responses API server that returns the same assistant message for every request.
+pub async fn create_mock_responses_server_repeating_assistant(message: &str) -> MockServer {
+    let server = responses::start_mock_server().await;
+    let body = responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_assistant_message("msg-1", message),
+        responses::ev_completed("resp-1"),
+    ]);
+    Mock::given(method("POST"))
+        .and(path_regex(".*/responses$"))
+        .respond_with(responses::sse_response(body))
+        .mount(&server)
+        .await;
+    server
+}

codex-rs/app-server/tests/common/responses.rs

@@ -1,3 +1,4 @@
+use core_test_support::responses;
 use serde_json::json;
 use std::path::Path;
 
@@ -14,85 +15,30 @@ pub fn create_shell_command_sse_response(
         "workdir": workdir.map(|w| w.to_string_lossy()),
         "timeout_ms": timeout_ms
     }))?;
-    let tool_call = json!({
-        "choices": [
-            {
-                "delta": {
-                    "tool_calls": [
-                        {
-                            "id": call_id,
-                            "function": {
-                                "name": "shell_command",
-                                "arguments": tool_call_arguments
-                            }
-                        }
-                    ]
-                },
-                "finish_reason": "tool_calls"
-            }
-        ]
-    });
-
-    let sse = format!(
-        "data: {}\n\ndata: DONE\n\n",
-        serde_json::to_string(&tool_call)?
-    );
-    Ok(sse)
+    Ok(responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_function_call(call_id, "shell_command", &tool_call_arguments),
+        responses::ev_completed("resp-1"),
+    ]))
 }
 
 pub fn create_final_assistant_message_sse_response(message: &str) -> anyhow::Result<String> {
-    let assistant_message = json!({
-        "choices": [
-            {
-                "delta": {
-                    "content": message
-                },
-                "finish_reason": "stop"
-            }
-        ]
-    });
-
-    let sse = format!(
-        "data: {}\n\ndata: DONE\n\n",
-        serde_json::to_string(&assistant_message)?
-    );
-    Ok(sse)
+    Ok(responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_assistant_message("msg-1", message),
+        responses::ev_completed("resp-1"),
+    ]))
 }
 
 pub fn create_apply_patch_sse_response(
     patch_content: &str,
     call_id: &str,
 ) -> anyhow::Result<String> {
-    // Use shell_command to call apply_patch with heredoc format
-    let command = format!("apply_patch <<'EOF'\n{patch_content}\nEOF");
-    let tool_call_arguments = serde_json::to_string(&json!({
-        "command": command
-    }))?;
-
-    let tool_call = json!({
-        "choices": [
-            {
-                "delta": {
-                    "tool_calls": [
-                        {
-                            "id": call_id,
-                            "function": {
-                                "name": "shell_command",
-                                "arguments": tool_call_arguments
-                            }
-                        }
-                    ]
-                },
-                "finish_reason": "tool_calls"
-            }
-        ]
-    });
-
-    let sse = format!(
-        "data: {}\n\ndata: DONE\n\n",
-        serde_json::to_string(&tool_call)?
-    );
-    Ok(sse)
+    Ok(responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_apply_patch_shell_command_call_via_heredoc(call_id, patch_content),
+        responses::ev_completed("resp-1"),
+    ]))
 }
 
 pub fn create_exec_command_sse_response(call_id: &str) -> anyhow::Result<String> {
@@ -108,28 +54,9 @@ pub fn create_exec_command_sse_response(call_id: &str) -> anyhow::Result<String>
         "cmd": command.join(" "),
         "yield_time_ms": 500
     }))?;
-    let tool_call = json!({
-        "choices": [
-            {
-                "delta": {
-                    "tool_calls": [
-                        {
-                            "id": call_id,
-                            "function": {
-                                "name": "exec_command",
-                                "arguments": tool_call_arguments
-                            }
-                        }
-                    ]
-                },
-                "finish_reason": "tool_calls"
-            }
-        ]
-    });
-
-    let sse = format!(
-        "data: {}\n\ndata: DONE\n\n",
-        serde_json::to_string(&tool_call)?
-    );
-    Ok(sse)
+    Ok(responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_function_call(call_id, "exec_command", &tool_call_arguments),
+        responses::ev_completed("resp-1"),
+    ]))
 }

codex-rs/app-server/tests/suite/auth.rs

@@ -37,7 +37,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "http://127.0.0.1:0/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 {requires_line}

codex-rs/app-server/tests/suite/codex_message_processor_flow.rs

@@ -1,7 +1,7 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
 use app_test_support::create_final_assistant_message_sse_response;
-use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_mock_responses_server_sequence;
 use app_test_support::create_shell_command_sse_response;
 use app_test_support::format_with_current_shell;
 use app_test_support::to_response;
@@ -65,7 +65,7 @@ async fn test_codex_jsonrpc_conversation_flow() -> Result<()> {
         )?,
         create_final_assistant_message_sse_response("Enjoy your new git repo!")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri())?;
 
     // Start MCP server and initialize.
@@ -197,7 +197,7 @@ async fn test_send_user_turn_changes_approval_policy_behavior() -> Result<()> {
         )?,
         create_final_assistant_message_sse_response("done 2")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri())?;
 
     // Start MCP server and initialize.
@@ -283,7 +283,7 @@ async fn test_send_user_turn_changes_approval_policy_behavior() -> Result<()> {
     )
     .await?;
 
-    // Wait for first TaskComplete
+    // Wait for first TurnComplete
     let _ = timeout(
         DEFAULT_READ_TIMEOUT,
         mcp.read_stream_until_notification_message("codex/event/task_complete"),
@@ -363,7 +363,7 @@ async fn test_send_user_turn_updates_sandbox_and_cwd_between_turns() -> Result<(
         )?,
         create_final_assistant_message_sse_response("done second")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri())?;
 
     let mut mcp = McpProcess::new(&codex_home).await?;
@@ -430,6 +430,7 @@ async fn test_send_user_turn_updates_sandbox_and_cwd_between_turns() -> Result<(
         mcp.read_stream_until_notification_message("codex/event/task_complete"),
     )
     .await??;
+    mcp.clear_message_buffer();
 
     let second_turn_id = mcp
         .send_send_user_turn_request(SendUserTurnParams {
@@ -499,7 +500,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

codex-rs/app-server/tests/suite/create_thread.rs

@@ -1,7 +1,6 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
 use app_test_support::create_final_assistant_message_sse_response;
-use app_test_support::create_mock_chat_completions_server;
 use app_test_support::to_response;
 use codex_app_server_protocol::AddConversationListenerParams;
 use codex_app_server_protocol::AddConversationSubscriptionResponse;
@@ -12,6 +11,7 @@ use codex_app_server_protocol::NewConversationResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::SendUserMessageParams;
 use codex_app_server_protocol::SendUserMessageResponse;
+use core_test_support::responses;
 use pretty_assertions::assert_eq;
 use serde_json::json;
 use std::path::Path;
@@ -23,8 +23,9 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_conversation_create_and_send_message_ok() -> Result<()> {
     // Mock server – we won't strictly rely on it, but provide one to satisfy any model wiring.
-    let responses = vec![create_final_assistant_message_sse_response("Done")?];
-    let server = create_mock_chat_completions_server(responses).await;
+    let response_body = create_final_assistant_message_sse_response("Done")?;
+    let server = responses::start_mock_server().await;
+    let response_mock = responses::mount_sse_sequence(&server, vec![response_body]).await;
 
     // Temporary Codex home with config pointing at the mock server.
     let codex_home = TempDir::new()?;
@@ -86,32 +87,30 @@ async fn test_conversation_create_and_send_message_ok() -> Result<()> {
     .await??;
     let _ok: SendUserMessageResponse = to_response::<SendUserMessageResponse>(send_resp)?;
 
-    // avoid race condition by waiting for the mock server to receive the chat.completions request
+    // Avoid race condition by waiting for the mock server to receive the responses request.
     let deadline = std::time::Instant::now() + DEFAULT_READ_TIMEOUT;
     let requests = loop {
-        let requests = server.received_requests().await.unwrap_or_default();
+        let requests = response_mock.requests();
         if !requests.is_empty() {
             break requests;
         }
         if std::time::Instant::now() >= deadline {
-            panic!("mock server did not receive the chat.completions request in time");
+            panic!("mock server did not receive the responses request in time");
         }
         tokio::time::sleep(std::time::Duration::from_millis(10)).await;
     };
 
-    // Verify the outbound request body matches expectations for Chat Completions.
+    // Verify the outbound request body matches expectations for Responses.
     let request = requests
         .first()
         .expect("mock server should have received at least one request");
-    let body = request.body_json::<serde_json::Value>()?;
+    let body = request.body_json();
     assert_eq!(body["model"], json!("o3"));
-    assert!(body["stream"].as_bool().unwrap_or(false));
-    let messages = body["messages"]
-        .as_array()
-        .expect("messages should be array");
-    let last = messages.last().expect("at least one message");
-    assert_eq!(last["role"], json!("user"));
-    assert_eq!(last["content"], json!("Hello"));
+    let user_texts = request.message_input_texts("user");
+    assert!(
+        user_texts.iter().any(|text| text == "Hello"),
+        "expected user input to include Hello, got {user_texts:?}"
+    );
 
     drop(server);
     Ok(())
@@ -133,7 +132,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

codex-rs/app-server/tests/suite/fork_thread.rs

@@ -0,0 +1,140 @@
+use anyhow::Result;
+use app_test_support::McpProcess;
+use app_test_support::create_fake_rollout;
+use app_test_support::to_response;
+use codex_app_server_protocol::ForkConversationParams;
+use codex_app_server_protocol::ForkConversationResponse;
+use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::NewConversationParams; // reused for overrides shape
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::SessionConfiguredNotification;
+use codex_core::protocol::EventMsg;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn fork_conversation_creates_new_rollout() -> Result<()> {
+    let codex_home = TempDir::new()?;
+
+    let preview = "Hello A";
+    let conversation_id = create_fake_rollout(
+        codex_home.path(),
+        "2025-01-02T12-00-00",
+        "2025-01-02T12:00:00Z",
+        preview,
+        Some("openai"),
+        None,
+    )?;
+
+    let original_path = codex_home
+        .path()
+        .join("sessions")
+        .join("2025")
+        .join("01")
+        .join("02")
+        .join(format!(
+            "rollout-2025-01-02T12-00-00-{conversation_id}.jsonl"
+        ));
+    assert!(
+        original_path.exists(),
+        "expected original rollout to exist at {}",
+        original_path.display()
+    );
+    let original_contents = std::fs::read_to_string(&original_path)?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let fork_req_id = mcp
+        .send_fork_conversation_request(ForkConversationParams {
+            path: Some(original_path.clone()),
+            conversation_id: None,
+            overrides: Some(NewConversationParams {
+                model: Some("o3".to_string()),
+                ..Default::default()
+            }),
+        })
+        .await?;
+
+    // Expect a sessionConfigured notification for the forked session.
+    let notification: JSONRPCNotification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("sessionConfigured"),
+    )
+    .await??;
+    let session_configured: ServerNotification = notification.try_into()?;
+    let ServerNotification::SessionConfigured(SessionConfiguredNotification {
+        model,
+        session_id,
+        rollout_path,
+        initial_messages: session_initial_messages,
+        ..
+    }) = session_configured
+    else {
+        unreachable!("expected sessionConfigured notification");
+    };
+
+    assert_eq!(model, "o3");
+    assert_ne!(
+        session_id.to_string(),
+        conversation_id,
+        "expected a new conversation id when forking"
+    );
+    assert_ne!(
+        rollout_path, original_path,
+        "expected a new rollout path when forking"
+    );
+    assert!(
+        rollout_path.exists(),
+        "expected forked rollout to exist at {}",
+        rollout_path.display()
+    );
+
+    let session_initial_messages =
+        session_initial_messages.expect("expected initial messages when forking from rollout");
+    match session_initial_messages.as_slice() {
+        [EventMsg::UserMessage(message)] => {
+            assert_eq!(message.message, preview);
+        }
+        other => panic!("unexpected initial messages from rollout fork: {other:#?}"),
+    }
+
+    // Then the response for forkConversation.
+    let fork_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(fork_req_id)),
+    )
+    .await??;
+    let ForkConversationResponse {
+        conversation_id: forked_id,
+        model: forked_model,
+        initial_messages: response_initial_messages,
+        rollout_path: response_rollout_path,
+    } = to_response::<ForkConversationResponse>(fork_resp)?;
+
+    assert_eq!(forked_model, "o3");
+    assert_eq!(response_rollout_path, rollout_path);
+    assert_ne!(forked_id.to_string(), conversation_id);
+
+    let response_initial_messages =
+        response_initial_messages.expect("expected initial messages in fork response");
+    match response_initial_messages.as_slice() {
+        [EventMsg::UserMessage(message)] => {
+            assert_eq!(message.message, preview);
+        }
+        other => panic!("unexpected initial messages in fork response: {other:#?}"),
+    }
+
+    let after_contents = std::fs::read_to_string(&original_path)?;
+    assert_eq!(
+        after_contents, original_contents,
+        "fork should not mutate the original rollout file"
+    );
+
+    Ok(())
+}

codex-rs/app-server/tests/suite/interrupt.rs

@@ -18,7 +18,7 @@ use tempfile::TempDir;
 use tokio::time::timeout;
 
 use app_test_support::McpProcess;
-use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_mock_responses_server_sequence;
 use app_test_support::create_shell_command_sse_response;
 use app_test_support::to_response;
 
@@ -56,7 +56,7 @@ async fn shell_command_interruption() -> anyhow::Result<()> {
     std::fs::create_dir(&working_directory)?;
 
     // Create mock server with a single SSE response: the long sleep command
-    let server = create_mock_chat_completions_server(vec![create_shell_command_sse_response(
+    let server = create_mock_responses_server_sequence(vec![create_shell_command_sse_response(
         shell_command.clone(),
         Some(&working_directory),
         Some(10_000), // 10 seconds timeout in ms
@@ -153,7 +153,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

codex-rs/app-server/tests/suite/login.rs

@@ -32,7 +32,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "http://127.0.0.1:0/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#,

codex-rs/app-server/tests/suite/mod.rs

@@ -3,6 +3,7 @@ mod auth;
 mod codex_message_processor_flow;
 mod config;
 mod create_thread;
+mod fork_thread;
 mod fuzzy_file_search;
 mod interrupt;
 mod list_resume;

codex-rs/app-server/tests/suite/send_message.rs

@@ -1,7 +1,5 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
-use app_test_support::create_final_assistant_message_sse_response;
-use app_test_support::create_mock_chat_completions_server;
 use app_test_support::to_response;
 use codex_app_server_protocol::AddConversationListenerParams;
 use codex_app_server_protocol::AddConversationSubscriptionResponse;
@@ -17,6 +15,7 @@ use codex_protocol::ThreadId;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::RawResponseItemEvent;
+use core_test_support::responses;
 use pretty_assertions::assert_eq;
 use std::path::Path;
 use tempfile::TempDir;
@@ -26,13 +25,21 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs
 
 #[tokio::test]
 async fn test_send_message_success() -> Result<()> {
-    // Spin up a mock completions server that immediately ends the Codex turn.
+    // Spin up a mock responses server that immediately ends the Codex turn.
     // Two Codex turns hit the mock model (session start + send-user-message). Provide two SSE responses.
-    let responses = vec![
-        create_final_assistant_message_sse_response("Done")?,
-        create_final_assistant_message_sse_response("Done")?,
-    ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = responses::start_mock_server().await;
+    let body1 = responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_assistant_message("msg-1", "Done"),
+        responses::ev_completed("resp-1"),
+    ]);
+    let body2 = responses::sse(vec![
+        responses::ev_response_created("resp-2"),
+        responses::ev_assistant_message("msg-2", "Done"),
+        responses::ev_completed("resp-2"),
+    ]);
+    let _response_mock1 = responses::mount_sse_once(&server, body1).await;
+    let _response_mock2 = responses::mount_sse_once(&server, body2).await;
 
     // Create a temporary Codex home with config pointing at the mock server.
     let codex_home = TempDir::new()?;
@@ -135,8 +142,13 @@ async fn send_message(
 
 #[tokio::test]
 async fn test_send_message_raw_notifications_opt_in() -> Result<()> {
-    let responses = vec![create_final_assistant_message_sse_response("Done")?];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = responses::start_mock_server().await;
+    let body = responses::sse(vec![
+        responses::ev_response_created("resp-1"),
+        responses::ev_assistant_message("msg-1", "Done"),
+        responses::ev_completed("resp-1"),
+    ]);
+    let _response_mock = responses::mount_sse_once(&server, body).await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -259,7 +271,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#
@@ -269,6 +281,7 @@ stream_max_retries = 0
 
 #[expect(clippy::expect_used)]
 async fn read_raw_response_item(mcp: &mut McpProcess, conversation_id: ThreadId) -> ResponseItem {
+    // TODO: Switch to rawResponseItem/completed once we migrate to app server v2 in codex web.
     loop {
         let raw_notification: JSONRPCNotification = timeout(
             DEFAULT_READ_TIMEOUT,

codex-rs/app-server/tests/suite/user_agent.rs

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use app_test_support::DEFAULT_CLIENT_NAME;
 use app_test_support::McpProcess;
 use app_test_support::to_response;
 use codex_app_server_protocol::GetUserAgentResponse;
@@ -25,13 +26,13 @@ async fn get_user_agent_returns_current_codex_user_agent() -> Result<()> {
     .await??;
 
     let os_info = os_info::get();
-    let originator = codex_core::default_client::originator().value.as_str();
+    let originator = DEFAULT_CLIENT_NAME;
     let os_type = os_info.os_type();
     let os_version = os_info.version();
     let architecture = os_info.architecture().unwrap_or("unknown");
     let terminal_ua = codex_core::terminal::user_agent();
     let user_agent = format!(
-        "{originator}/0.0.0 ({os_type} {os_version}; {architecture}) {terminal_ua} (codex-app-server-tests; 0.1.0)"
+        "{originator}/0.0.0 ({os_type} {os_version}; {architecture}) {terminal_ua} ({DEFAULT_CLIENT_NAME}; 0.1.0)"
     );
 
     let received: GetUserAgentResponse = to_response(response)?;

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

@@ -67,7 +67,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "http://127.0.0.1:0/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 {requires_line}

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

@@ -0,0 +1,137 @@
+use anyhow::Result;
+use app_test_support::McpProcess;
+use app_test_support::create_mock_responses_server_sequence_unchecked;
+use app_test_support::to_response;
+use codex_app_server_protocol::ClientInfo;
+use codex_app_server_protocol::InitializeResponse;
+use codex_app_server_protocol::JSONRPCMessage;
+use pretty_assertions::assert_eq;
+use std::path::Path;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test]
+async fn initialize_uses_client_info_name_as_originator() -> Result<()> {
+    let responses = Vec::new();
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+
+    let message = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_client_info(ClientInfo {
+            name: "codex_vscode".to_string(),
+            title: Some("Codex VS Code Extension".to_string()),
+            version: "0.1.0".to_string(),
+        }),
+    )
+    .await??;
+
+    let JSONRPCMessage::Response(response) = message else {
+        anyhow::bail!("expected initialize response, got {message:?}");
+    };
+    let InitializeResponse { user_agent } = to_response::<InitializeResponse>(response)?;
+
+    assert!(user_agent.starts_with("codex_vscode/"));
+    Ok(())
+}
+
+#[tokio::test]
+async fn initialize_respects_originator_override_env_var() -> Result<()> {
+    let responses = Vec::new();
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+    let mut mcp = McpProcess::new_with_env(
+        codex_home.path(),
+        &[(
+            "CODEX_INTERNAL_ORIGINATOR_OVERRIDE",
+            Some("codex_originator_via_env_var"),
+        )],
+    )
+    .await?;
+
+    let message = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_client_info(ClientInfo {
+            name: "codex_vscode".to_string(),
+            title: Some("Codex VS Code Extension".to_string()),
+            version: "0.1.0".to_string(),
+        }),
+    )
+    .await??;
+
+    let JSONRPCMessage::Response(response) = message else {
+        anyhow::bail!("expected initialize response, got {message:?}");
+    };
+    let InitializeResponse { user_agent } = to_response::<InitializeResponse>(response)?;
+
+    assert!(user_agent.starts_with("codex_originator_via_env_var/"));
+    Ok(())
+}
+
+#[tokio::test]
+async fn initialize_rejects_invalid_client_name() -> Result<()> {
+    let responses = Vec::new();
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+    let mut mcp = McpProcess::new_with_env(
+        codex_home.path(),
+        &[("CODEX_INTERNAL_ORIGINATOR_OVERRIDE", None)],
+    )
+    .await?;
+
+    let message = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_client_info(ClientInfo {
+            name: "bad\rname".to_string(),
+            title: Some("Bad Client".to_string()),
+            version: "0.1.0".to_string(),
+        }),
+    )
+    .await??;
+
+    let JSONRPCMessage::Error(error) = message else {
+        anyhow::bail!("expected initialize error, got {message:?}");
+    };
+
+    assert_eq!(error.error.code, -32600);
+    assert_eq!(
+        error.error.message,
+        "Invalid clientInfo.name: 'bad\rname'. Must be a valid HTTP header value."
+    );
+    assert_eq!(error.error.data, None);
+    Ok(())
+}
+
+// Helper to create a config.toml pointing at the mock model server.
+fn create_config_toml(
+    codex_home: &Path,
+    server_uri: &str,
+    approval_policy: &str,
+) -> std::io::Result<()> {
+    let config_toml = codex_home.join("config.toml");
+    std::fs::write(
+        config_toml,
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "{approval_policy}"
+sandbox_mode = "read-only"
+
+model_provider = "mock_provider"
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{server_uri}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )
+}

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

@@ -1,11 +1,14 @@
 mod account;
 mod config_rpc;
+mod initialize;
 mod model_list;
 mod output_schema;
 mod rate_limits;
 mod review;
 mod thread_archive;
+mod thread_fork;
 mod thread_list;
+mod thread_loaded_list;
 mod thread_resume;
 mod thread_rollback;
 mod thread_start;

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

@@ -1,7 +1,6 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
-use app_test_support::create_final_assistant_message_sse_response;
-use app_test_support::create_mock_chat_completions_server_unchecked;
+use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
 use codex_app_server_protocol::ItemCompletedNotification;
 use codex_app_server_protocol::ItemStartedNotification;
@@ -44,10 +43,7 @@ async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()
         "overall_confidence_score": 0.75
     })
     .to_string();
-    let responses = vec![create_final_assistant_message_sse_response(
-        &review_payload,
-    )?];
-    let server = create_mock_chat_completions_server_unchecked(responses).await;
+    let server = create_mock_responses_server_repeating_assistant(&review_payload).await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -135,7 +131,7 @@ async fn review_start_runs_review_turn_and_emits_code_review_item() -> Result<()
 
 #[tokio::test]
 async fn review_start_rejects_empty_base_branch() -> Result<()> {
-    let server = create_mock_chat_completions_server_unchecked(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -176,10 +172,7 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<(
         "overall_confidence_score": 0.5
     })
     .to_string();
-    let responses = vec![create_final_assistant_message_sse_response(
-        &review_payload,
-    )?];
-    let server = create_mock_chat_completions_server_unchecked(responses).await;
+    let server = create_mock_responses_server_repeating_assistant(&review_payload).await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -219,7 +212,7 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<(
 
 #[tokio::test]
 async fn review_start_rejects_empty_commit_sha() -> Result<()> {
-    let server = create_mock_chat_completions_server_unchecked(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -254,7 +247,7 @@ async fn review_start_rejects_empty_commit_sha() -> Result<()> {
 
 #[tokio::test]
 async fn review_start_rejects_empty_custom_instructions() -> Result<()> {
-    let server = create_mock_chat_completions_server_unchecked(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -320,7 +313,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

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

@@ -0,0 +1,140 @@
+use anyhow::Result;
+use app_test_support::McpProcess;
+use app_test_support::create_fake_rollout;
+use app_test_support::create_mock_responses_server_repeating_assistant;
+use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCNotification;
+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::ThreadStartedNotification;
+use codex_app_server_protocol::TurnStatus;
+use codex_app_server_protocol::UserInput;
+use pretty_assertions::assert_eq;
+use std::path::Path;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test]
+async fn thread_fork_creates_new_thread_and_emits_started() -> 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 preview = "Saved user message";
+    let conversation_id = create_fake_rollout(
+        codex_home.path(),
+        "2025-01-05T12-00-00",
+        "2025-01-05T12:00:00Z",
+        preview,
+        Some("mock_provider"),
+        None,
+    )?;
+
+    let original_path = codex_home
+        .path()
+        .join("sessions")
+        .join("2025")
+        .join("01")
+        .join("05")
+        .join(format!(
+            "rollout-2025-01-05T12-00-00-{conversation_id}.jsonl"
+        ));
+    assert!(
+        original_path.exists(),
+        "expected original rollout to exist at {}",
+        original_path.display()
+    );
+    let original_contents = std::fs::read_to_string(&original_path)?;
+
+    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, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
+
+    let after_contents = std::fs::read_to_string(&original_path)?;
+    assert_eq!(
+        after_contents, original_contents,
+        "fork should not mutate the original rollout file"
+    );
+
+    assert_ne!(thread.id, conversation_id);
+    assert_eq!(thread.preview, preview);
+    assert_eq!(thread.model_provider, "mock_provider");
+    assert!(thread.path.is_absolute());
+    assert_ne!(thread.path, original_path);
+    assert!(thread.cwd.is_absolute());
+    assert_eq!(thread.source, SessionSource::VsCode);
+
+    assert_eq!(
+        thread.turns.len(),
+        1,
+        "expected forked thread to include one turn"
+    );
+    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()
+                }]
+            );
+        }
+        other => panic!("expected user message item, got {other:?}"),
+    }
+
+    // A corresponding thread/started notification should arrive.
+    let notif: JSONRPCNotification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("thread/started"),
+    )
+    .await??;
+    let started: ThreadStartedNotification =
+        serde_json::from_value(notif.params.expect("params must be present"))?;
+    assert_eq!(started.thread, thread);
+
+    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");
+    std::fs::write(
+        config_toml,
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "never"
+sandbox_mode = "read-only"
+
+model_provider = "mock_provider"
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{server_uri}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )
+}

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

@@ -0,0 +1,139 @@
+use anyhow::Result;
+use app_test_support::McpProcess;
+use app_test_support::create_mock_responses_server_repeating_assistant;
+use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ThreadLoadedListParams;
+use codex_app_server_protocol::ThreadLoadedListResponse;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use pretty_assertions::assert_eq;
+use std::path::Path;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test]
+async fn thread_loaded_list_returns_loaded_thread_ids() -> 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 mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let thread_id = start_thread(&mut mcp).await?;
+
+    let list_id = mcp
+        .send_thread_loaded_list_request(ThreadLoadedListParams::default())
+        .await?;
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
+    )
+    .await??;
+    let ThreadLoadedListResponse {
+        mut data,
+        next_cursor,
+    } = to_response::<ThreadLoadedListResponse>(resp)?;
+    data.sort();
+    assert_eq!(data, vec![thread_id]);
+    assert_eq!(next_cursor, None);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn thread_loaded_list_paginates() -> 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 mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let first = start_thread(&mut mcp).await?;
+    let second = start_thread(&mut mcp).await?;
+
+    let mut expected = [first, second];
+    expected.sort();
+
+    let list_id = mcp
+        .send_thread_loaded_list_request(ThreadLoadedListParams {
+            cursor: None,
+            limit: Some(1),
+        })
+        .await?;
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
+    )
+    .await??;
+    let ThreadLoadedListResponse {
+        data: first_page,
+        next_cursor,
+    } = to_response::<ThreadLoadedListResponse>(resp)?;
+    assert_eq!(first_page, vec![expected[0].clone()]);
+    assert_eq!(next_cursor, Some(expected[0].clone()));
+
+    let list_id = mcp
+        .send_thread_loaded_list_request(ThreadLoadedListParams {
+            cursor: next_cursor,
+            limit: Some(1),
+        })
+        .await?;
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
+    )
+    .await??;
+    let ThreadLoadedListResponse {
+        data: second_page,
+        next_cursor,
+    } = to_response::<ThreadLoadedListResponse>(resp)?;
+    assert_eq!(second_page, vec![expected[1].clone()]);
+    assert_eq!(next_cursor, None);
+
+    Ok(())
+}
+
+fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
+    let config_toml = codex_home.join("config.toml");
+    std::fs::write(
+        config_toml,
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "never"
+sandbox_mode = "read-only"
+
+model_provider = "mock_provider"
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{server_uri}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )
+}
+
+async fn start_thread(mcp: &mut McpProcess) -> Result<String> {
+    let req_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("gpt-5.1".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
+    Ok(thread.id)
+}

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

@@ -1,7 +1,7 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
 use app_test_support::create_fake_rollout;
-use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
@@ -23,7 +23,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs
 
 #[tokio::test]
 async fn thread_resume_returns_original_thread() -> Result<()> {
-    let server = create_mock_chat_completions_server(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -66,7 +66,7 @@ async fn thread_resume_returns_original_thread() -> Result<()> {
 
 #[tokio::test]
 async fn thread_resume_returns_rollout_history() -> Result<()> {
-    let server = create_mock_chat_completions_server(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -130,7 +130,7 @@ async fn thread_resume_returns_rollout_history() -> Result<()> {
 
 #[tokio::test]
 async fn thread_resume_prefers_path_over_thread_id() -> Result<()> {
-    let server = create_mock_chat_completions_server(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -174,7 +174,7 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> {
 
 #[tokio::test]
 async fn thread_resume_supports_history_and_overrides() -> Result<()> {
-    let server = create_mock_chat_completions_server(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
 
@@ -247,7 +247,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

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

@@ -1,7 +1,7 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
 use app_test_support::create_final_assistant_message_sse_response;
-use app_test_support::create_mock_chat_completions_server_unchecked;
+use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::to_response;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
@@ -28,7 +28,7 @@ async fn thread_rollback_drops_last_turns_and_persists_to_rollout() -> Result<()
         create_final_assistant_message_sse_response("Done")?,
         create_final_assistant_message_sse_response("Done")?,
     ];
-    let server = create_mock_chat_completions_server_unchecked(responses).await;
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -168,7 +168,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

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

@@ -1,6 +1,6 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
-use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
 use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCResponse;
@@ -17,7 +17,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs
 #[tokio::test]
 async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
     // Provide a mock server and config so model wiring is valid.
-    let server = create_mock_chat_completions_server(vec![]).await;
+    let server = create_mock_responses_server_repeating_assistant("Done").await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri())?;
@@ -85,7 +85,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

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

@@ -2,7 +2,7 @@
 
 use anyhow::Result;
 use app_test_support::McpProcess;
-use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_mock_responses_server_sequence;
 use app_test_support::create_shell_command_sse_response;
 use app_test_support::to_response;
 use codex_app_server_protocol::JSONRPCNotification;
@@ -41,7 +41,7 @@ async fn turn_interrupt_aborts_running_turn() -> Result<()> {
     std::fs::create_dir(&working_directory)?;
 
     // Mock server: long-running shell command then (after abort) nothing else needed.
-    let server = create_mock_chat_completions_server(vec![create_shell_command_sse_response(
+    let server = create_mock_responses_server_sequence(vec![create_shell_command_sse_response(
         shell_command.clone(),
         Some(&working_directory),
         Some(10_000),
@@ -135,7 +135,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

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

@@ -3,11 +3,12 @@ use app_test_support::McpProcess;
 use app_test_support::create_apply_patch_sse_response;
 use app_test_support::create_exec_command_sse_response;
 use app_test_support::create_final_assistant_message_sse_response;
-use app_test_support::create_mock_chat_completions_server;
-use app_test_support::create_mock_chat_completions_server_unchecked;
+use app_test_support::create_mock_responses_server_sequence;
+use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::create_shell_command_sse_response;
 use app_test_support::format_with_current_shell_display;
 use app_test_support::to_response;
+use codex_app_server_protocol::ClientInfo;
 use codex_app_server_protocol::CommandExecutionApprovalDecision;
 use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
 use codex_app_server_protocol::CommandExecutionStatus;
@@ -40,6 +41,76 @@ use tempfile::TempDir;
 use tokio::time::timeout;
 
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+const TEST_ORIGINATOR: &str = "codex_vscode";
+
+#[tokio::test]
+async fn turn_start_sends_originator_header() -> Result<()> {
+    let responses = vec![create_final_assistant_message_sse_response("Done")?];
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_client_info(ClientInfo {
+            name: TEST_ORIGINATOR.to_string(),
+            title: Some("Codex VS Code Extension".to_string()),
+            version: "0.1.0".to_string(),
+        }),
+    )
+    .await??;
+
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
+
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: "Hello".to_string(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    let requests = server
+        .received_requests()
+        .await
+        .expect("failed to fetch received requests");
+    assert!(!requests.is_empty());
+    for request in requests {
+        let originator = request
+            .headers
+            .get("originator")
+            .expect("originator header missing");
+        assert_eq!(originator.to_str()?, TEST_ORIGINATOR);
+    }
+
+    Ok(())
+}
 
 #[tokio::test]
 async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<()> {
@@ -50,7 +121,7 @@ async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<(
         create_final_assistant_message_sse_response("Done")?,
         create_final_assistant_message_sse_response("Done")?,
     ];
-    let server = create_mock_chat_completions_server_unchecked(responses).await;
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri(), "never")?;
@@ -157,7 +228,7 @@ async fn turn_start_accepts_local_image_input() -> Result<()> {
     ];
     // Use the unchecked variant because the request payload includes a LocalImage
     // which the strict matcher does not currently cover.
-    let server = create_mock_chat_completions_server_unchecked(responses).await;
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
 
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri(), "never")?;
@@ -233,7 +304,7 @@ async fn turn_start_exec_approval_toggle_v2() -> Result<()> {
         )?,
         create_final_assistant_message_sse_response("done 2")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     // Default approval is untrusted to force elicitation on first turn.
     create_config_toml(codex_home.as_path(), &server.uri(), "untrusted")?;
 
@@ -357,7 +428,7 @@ async fn turn_start_exec_approval_decline_v2() -> Result<()> {
         )?,
         create_final_assistant_message_sse_response("done")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(codex_home.as_path(), &server.uri(), "untrusted")?;
 
     let mut mcp = McpProcess::new(codex_home.as_path()).await?;
@@ -503,7 +574,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
         )?,
         create_final_assistant_message_sse_response("done second")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri(), "untrusted")?;
 
     let mut mcp = McpProcess::new(&codex_home).await?;
@@ -554,6 +625,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
         mcp.read_stream_until_notification_message("codex/event/task_complete"),
     )
     .await??;
+    mcp.clear_message_buffer();
 
     // second turn with workspace-write and second_cwd, ensure exec begins in second_cwd
     let second_turn = mcp
@@ -636,7 +708,7 @@ async fn turn_start_file_change_approval_v2() -> Result<()> {
         create_apply_patch_sse_response(patch, "patch-call")?,
         create_final_assistant_message_sse_response("patch applied")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri(), "untrusted")?;
 
     let mut mcp = McpProcess::new(&codex_home).await?;
@@ -812,7 +884,7 @@ async fn turn_start_file_change_approval_accept_for_session_persists_v2() -> Res
         create_apply_patch_sse_response(patch_2, "patch-call-2")?,
         create_final_assistant_message_sse_response("patch 2 applied")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri(), "untrusted")?;
 
     let mut mcp = McpProcess::new(&codex_home).await?;
@@ -986,7 +1058,7 @@ async fn turn_start_file_change_approval_decline_v2() -> Result<()> {
         create_apply_patch_sse_response(patch, "patch-call")?,
         create_final_assistant_message_sse_response("patch declined")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     create_config_toml(&codex_home, &server.uri(), "untrusted")?;
 
     let mut mcp = McpProcess::new(&codex_home).await?;
@@ -1124,7 +1196,7 @@ async fn command_execution_notifications_include_process_id() -> Result<()> {
         create_exec_command_sse_response("uexec-1")?,
         create_final_assistant_message_sse_response("done")?,
     ];
-    let server = create_mock_chat_completions_server(responses).await;
+    let server = create_mock_responses_server_sequence(responses).await;
     let codex_home = TempDir::new()?;
     create_config_toml(codex_home.path(), &server.uri(), "never")?;
     let config_toml = codex_home.path().join("config.toml");
@@ -1263,7 +1335,7 @@ model_provider = "mock_provider"
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
-wire_api = "chat"
+wire_api = "responses"
 request_max_retries = 0
 stream_max_retries = 0
 "#

codex-rs/apply-patch/BUILD.bazel

@@ -0,0 +1,11 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+exports_files(["apply_patch_tool_instructions.md"])
+
+codex_rust_crate(
+    name = "apply-patch",
+    crate_name = "codex_apply_patch",
+    compile_data = [
+        "apply_patch_tool_instructions.md",
+    ],
+)

codex-rs/apply-patch/tests/suite/scenarios.rs

@@ -1,3 +1,4 @@
+use codex_utils_cargo_bin::find_resource;
 use pretty_assertions::assert_eq;
 use std::collections::BTreeMap;
 use std::fs;
@@ -8,7 +9,7 @@ use tempfile::tempdir;
 
 #[test]
 fn test_apply_patch_scenarios() -> anyhow::Result<()> {
-    let scenarios_dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/scenarios");
+    let scenarios_dir = find_resource!("tests/fixtures/scenarios")?;
     for scenario in fs::read_dir(scenarios_dir)? {
         let scenario = scenario?;
         let path = scenario.path();

codex-rs/arg0/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "arg0",
+    crate_name = "codex_arg0",
+)

codex-rs/arg0/src/lib.rs

@@ -145,11 +145,41 @@ where
 /// that `apply_patch` can be on the PATH without requiring the user to
 /// install a separate `apply_patch` executable, simplifying the deployment of
 /// Codex CLI.
+/// Note: In debug builds the temp-dir guard is disabled to ease local testing.
 ///
 /// IMPORTANT: This function modifies the PATH environment variable, so it MUST
 /// be called before multiple threads are spawned.
 pub fn prepend_path_entry_for_codex_aliases() -> std::io::Result<TempDir> {
-    let temp_dir = TempDir::new()?;
+    let codex_home = codex_core::config::find_codex_home()?;
+    #[cfg(not(debug_assertions))]
+    {
+        // Guard against placing helpers in system temp directories outside debug builds.
+        let temp_root = std::env::temp_dir();
+        if codex_home.starts_with(&temp_root) {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                format!(
+                    "Refusing to create helper binaries under temporary dir {temp_root:?} (codex_home: {codex_home:?})"
+                ),
+            ));
+        }
+    }
+
+    std::fs::create_dir_all(&codex_home)?;
+    // Use a CODEX_HOME-scoped temp root to avoid cluttering the top-level directory.
+    let temp_root = codex_home.join("tmp").join("path");
+    std::fs::create_dir_all(&temp_root)?;
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+
+        // Ensure only the current user can access the temp directory.
+        std::fs::set_permissions(&temp_root, std::fs::Permissions::from_mode(0o700))?;
+    }
+
+    let temp_dir = tempfile::Builder::new()
+        .prefix("codex-arg0")
+        .tempdir_in(&temp_root)?;
     let path = temp_dir.path();
 
     for filename in &[

codex-rs/async-utils/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "async-utils",
+    crate_name = "codex_async_utils",
+)

codex-rs/backend-client/BUILD.bazel

@@ -0,0 +1,7 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "backend-client",
+    crate_name = "codex_backend_client",
+    compile_data = glob(["tests/fixtures/**"]),
+)

codex-rs/backend-client/src/client.rs

@@ -73,8 +73,8 @@ impl Client {
         })
     }
 
-    pub async fn from_auth(base_url: impl Into<String>, auth: &CodexAuth) -> Result<Self> {
-        let token = auth.get_token().await.map_err(anyhow::Error::from)?;
+    pub fn from_auth(base_url: impl Into<String>, auth: &CodexAuth) -> Result<Self> {
+        let token = auth.get_token().map_err(anyhow::Error::from)?;
         let mut client = Self::new(base_url)?
             .with_user_agent(get_codex_user_agent())
             .with_bearer_token(token);

codex-rs/chatgpt/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "chatgpt",
+    crate_name = "codex_chatgpt",
+)

codex-rs/chatgpt/Cargo.toml

@@ -12,6 +12,7 @@ anyhow = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 codex-common = { workspace = true, features = ["cli"] }
 codex-core = { workspace = true }
+codex-utils-cargo-bin = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["full"] }

codex-rs/chatgpt/src/chatgpt_token.rs

@@ -1,4 +1,4 @@
-use codex_core::CodexAuth;
+use codex_core::AuthManager;
 use std::path::Path;
 use std::sync::LazyLock;
 use std::sync::RwLock;
@@ -23,9 +23,10 @@ pub async fn init_chatgpt_token_from_auth(
     codex_home: &Path,
     auth_credentials_store_mode: AuthCredentialsStoreMode,
 ) -> std::io::Result<()> {
-    let auth = CodexAuth::from_auth_storage(codex_home, auth_credentials_store_mode)?;
-    if let Some(auth) = auth {
-        let token_data = auth.get_token_data().await?;
+    let auth_manager =
+        AuthManager::new(codex_home.to_path_buf(), false, auth_credentials_store_mode);
+    if let Some(auth) = auth_manager.auth().await {
+        let token_data = auth.get_token_data()?;
         set_chatgpt_token_data(token_data);
     }
     Ok(())

codex-rs/chatgpt/tests/suite/apply_command_e2e.rs

@@ -1,6 +1,6 @@
 use codex_chatgpt::apply_command::apply_diff_from_task;
 use codex_chatgpt::get_task::GetTaskResponse;
-use std::path::Path;
+use codex_utils_cargo_bin::find_resource;
 use tempfile::TempDir;
 use tokio::process::Command;
 
@@ -68,8 +68,8 @@ async fn create_temp_git_repo() -> anyhow::Result<TempDir> {
 }
 
 async fn mock_get_task_with_fixture() -> anyhow::Result<GetTaskResponse> {
-    let fixture_path = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/task_turn_fixture.json");
-    let fixture_content = std::fs::read_to_string(fixture_path)?;
+    let fixture_path = find_resource!("tests/task_turn_fixture.json")?;
+    let fixture_content = tokio::fs::read_to_string(fixture_path).await?;
     let response: GetTaskResponse = serde_json::from_str(&fixture_content)?;
     Ok(response)
 }

codex-rs/cli/BUILD.bazel

@@ -0,0 +1,10 @@
+load("//:defs.bzl", "codex_rust_crate", "multiplatform_binaries")
+
+codex_rust_crate(
+    name = "cli",
+    crate_name = "codex_cli",
+)
+
+multiplatform_binaries(
+    name = "codex",
+)

codex-rs/cli/Cargo.toml

@@ -30,7 +30,6 @@ codex-exec = { workspace = true }
 codex-execpolicy = { workspace = true }
 codex-login = { workspace = true }
 codex-mcp-server = { workspace = true }
-codex-process-hardening = { workspace = true }
 codex-protocol = { workspace = true }
 codex-responses-api-proxy = { workspace = true }
 codex-rmcp-client = { workspace = true }
@@ -38,7 +37,6 @@ codex-stdio-to-uds = { workspace = true }
 codex-tui = { workspace = true }
 codex-tui2 = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
-ctor = { workspace = true }
 libc = { workspace = true }
 owo-colors = { workspace = true }
 regex-lite = { workspace = true }

codex-rs/cli/src/login.rs

@@ -14,6 +14,18 @@ use std::io::IsTerminal;
 use std::io::Read;
 use std::path::PathBuf;
 
+const CHATGPT_LOGIN_DISABLED_MESSAGE: &str =
+    "ChatGPT login is disabled. Use API key login instead.";
+const API_KEY_LOGIN_DISABLED_MESSAGE: &str =
+    "API key login is disabled. Use ChatGPT login instead.";
+const LOGIN_SUCCESS_MESSAGE: &str = "Successfully logged in";
+
+fn print_login_server_start(actual_port: u16, auth_url: &str) {
+    eprintln!(
+        "Starting local login server on http://localhost:{actual_port}.\nIf your browser did not open, navigate to this URL to authenticate:\n\n{auth_url}"
+    );
+}
+
 pub async fn login_with_chatgpt(
     codex_home: PathBuf,
     forced_chatgpt_workspace_id: Option<String>,
@@ -27,10 +39,7 @@ pub async fn login_with_chatgpt(
     );
     let server = run_login_server(opts)?;
 
-    eprintln!(
-        "Starting local login server on http://localhost:{}.\nIf your browser did not open, navigate to this URL to authenticate:\n\n{}",
-        server.actual_port, server.auth_url,
-    );
+    print_login_server_start(server.actual_port, &server.auth_url);
 
     server.block_until_done().await
 }
@@ -39,7 +48,7 @@ pub async fn run_login_with_chatgpt(cli_config_overrides: CliConfigOverrides) ->
     let config = load_config_or_exit(cli_config_overrides).await;
 
     if matches!(config.forced_login_method, Some(ForcedLoginMethod::Api)) {
-        eprintln!("ChatGPT login is disabled. Use API key login instead.");
+        eprintln!("{CHATGPT_LOGIN_DISABLED_MESSAGE}");
         std::process::exit(1);
     }
 
@@ -53,7 +62,7 @@ pub async fn run_login_with_chatgpt(cli_config_overrides: CliConfigOverrides) ->
     .await
     {
         Ok(_) => {
-            eprintln!("Successfully logged in");
+            eprintln!("{LOGIN_SUCCESS_MESSAGE}");
             std::process::exit(0);
         }
         Err(e) => {
@@ -70,7 +79,7 @@ pub async fn run_login_with_api_key(
     let config = load_config_or_exit(cli_config_overrides).await;
 
     if matches!(config.forced_login_method, Some(ForcedLoginMethod::Chatgpt)) {
-        eprintln!("API key login is disabled. Use ChatGPT login instead.");
+        eprintln!("{API_KEY_LOGIN_DISABLED_MESSAGE}");
         std::process::exit(1);
     }
 
@@ -80,7 +89,7 @@ pub async fn run_login_with_api_key(
         config.cli_auth_credentials_store_mode,
     ) {
         Ok(_) => {
-            eprintln!("Successfully logged in");
+            eprintln!("{LOGIN_SUCCESS_MESSAGE}");
             std::process::exit(0);
         }
         Err(e) => {
@@ -125,7 +134,7 @@ pub async fn run_login_with_device_code(
 ) -> ! {
     let config = load_config_or_exit(cli_config_overrides).await;
     if matches!(config.forced_login_method, Some(ForcedLoginMethod::Api)) {
-        eprintln!("ChatGPT login is disabled. Use API key login instead.");
+        eprintln!("{CHATGPT_LOGIN_DISABLED_MESSAGE}");
         std::process::exit(1);
     }
     let forced_chatgpt_workspace_id = config.forced_chatgpt_workspace_id.clone();
@@ -140,7 +149,7 @@ pub async fn run_login_with_device_code(
     }
     match run_device_code_login(opts).await {
         Ok(()) => {
-            eprintln!("Successfully logged in");
+            eprintln!("{LOGIN_SUCCESS_MESSAGE}");
             std::process::exit(0);
         }
         Err(e) => {
@@ -150,12 +159,74 @@ pub async fn run_login_with_device_code(
     }
 }
 
+/// Prefers device-code login (with `open_browser = false`) when headless environment is detected, but keeps
+/// `codex login` working in environments where device-code may be disabled/feature-gated.
+/// If `run_device_code_login` returns `ErrorKind::NotFound` ("device-code unsupported"), this
+/// falls back to starting the local browser login server.
+pub async fn run_login_with_device_code_fallback_to_browser(
+    cli_config_overrides: CliConfigOverrides,
+    issuer_base_url: Option<String>,
+    client_id: Option<String>,
+) -> ! {
+    let config = load_config_or_exit(cli_config_overrides).await;
+    if matches!(config.forced_login_method, Some(ForcedLoginMethod::Api)) {
+        eprintln!("{CHATGPT_LOGIN_DISABLED_MESSAGE}");
+        std::process::exit(1);
+    }
+
+    let forced_chatgpt_workspace_id = config.forced_chatgpt_workspace_id.clone();
+    let mut opts = ServerOptions::new(
+        config.codex_home,
+        client_id.unwrap_or(CLIENT_ID.to_string()),
+        forced_chatgpt_workspace_id,
+        config.cli_auth_credentials_store_mode,
+    );
+    if let Some(iss) = issuer_base_url {
+        opts.issuer = iss;
+    }
+    opts.open_browser = false;
+
+    match run_device_code_login(opts.clone()).await {
+        Ok(()) => {
+            eprintln!("{LOGIN_SUCCESS_MESSAGE}");
+            std::process::exit(0);
+        }
+        Err(e) => {
+            if e.kind() == std::io::ErrorKind::NotFound {
+                eprintln!("Device code login is not enabled; falling back to browser login.");
+                match run_login_server(opts) {
+                    Ok(server) => {
+                        print_login_server_start(server.actual_port, &server.auth_url);
+                        match server.block_until_done().await {
+                            Ok(()) => {
+                                eprintln!("{LOGIN_SUCCESS_MESSAGE}");
+                                std::process::exit(0);
+                            }
+                            Err(e) => {
+                                eprintln!("Error logging in: {e}");
+                                std::process::exit(1);
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        eprintln!("Error logging in: {e}");
+                        std::process::exit(1);
+                    }
+                }
+            } else {
+                eprintln!("Error logging in with device code: {e}");
+                std::process::exit(1);
+            }
+        }
+    }
+}
+
 pub async fn run_login_status(cli_config_overrides: CliConfigOverrides) -> ! {
     let config = load_config_or_exit(cli_config_overrides).await;
 
     match CodexAuth::from_auth_storage(&config.codex_home, config.cli_auth_credentials_store_mode) {
         Ok(Some(auth)) => match auth.mode {
-            AuthMode::ApiKey => match auth.get_token().await {
+            AuthMode::ApiKey => match auth.get_token() {
                 Ok(api_key) => {
                     eprintln!("Logged in using an API key - {}", safe_format_key(&api_key));
                     std::process::exit(0);

codex-rs/cli/src/main.rs

@@ -14,9 +14,11 @@ use codex_cli::login::run_login_status;
 use codex_cli::login::run_login_with_api_key;
 use codex_cli::login::run_login_with_chatgpt;
 use codex_cli::login::run_login_with_device_code;
+use codex_cli::login::run_login_with_device_code_fallback_to_browser;
 use codex_cli::login::run_logout;
 use codex_cloud_tasks::Cli as CloudTasksCli;
 use codex_common::CliConfigOverrides;
+use codex_core::env::is_headless_environment;
 use codex_exec::Cli as ExecCli;
 use codex_exec::Command as ExecCommand;
 use codex_exec::ReviewArgs;
@@ -418,14 +420,6 @@ fn stage_str(stage: codex_core::features::Stage) -> &'static str {
     }
 }
 
-/// As early as possible in the process lifecycle, apply hardening measures. We
-/// skip this in debug builds to avoid interfering with debugging.
-#[ctor::ctor]
-#[cfg(not(debug_assertions))]
-fn pre_main_hardening() {
-    codex_process_hardening::pre_main_hardening();
-}
-
 fn main() -> anyhow::Result<()> {
     arg0_dispatch_or_else(|codex_linux_sandbox_exe| async move {
         cli_main(codex_linux_sandbox_exe).await?;
@@ -539,6 +533,13 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
                     } else if login_cli.with_api_key {
                         let api_key = read_api_key_from_stdin();
                         run_login_with_api_key(login_cli.config_overrides, api_key).await;
+                    } else if is_headless_environment() {
+                        run_login_with_device_code_fallback_to_browser(
+                            login_cli.config_overrides,
+                            login_cli.issuer_base_url,
+                            login_cli.client_id,
+                        )
+                        .await;
                     } else {
                         run_login_with_chatgpt(login_cli.config_overrides).await;
                     }

codex-rs/cloud-tasks-client/BUILD.bazel

@@ -0,0 +1,10 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "cloud-tasks-client",
+    crate_name = "codex_cloud_tasks_client",
+    crate_features = [
+        "mock",
+        "online",
+    ],
+)

codex-rs/cloud-tasks/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "cloud-tasks",
+    crate_name = "codex_cloud_tasks",
+)

codex-rs/cloud-tasks/src/lib.rs

@@ -10,7 +10,6 @@ pub use cli::Cli;
 use anyhow::anyhow;
 use chrono::Utc;
 use codex_cloud_tasks_client::TaskStatus;
-use codex_login::AuthManager;
 use owo_colors::OwoColorize;
 use owo_colors::Stream;
 use std::cmp::Ordering;
@@ -65,7 +64,11 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result<BackendContext>
     append_error_log(format!("startup: base_url={base_url} path_style={style}"));
 
     let auth_manager = util::load_auth_manager().await;
-    let auth = match auth_manager.as_ref().and_then(AuthManager::auth) {
+    let auth = match auth_manager.as_ref() {
+        Some(manager) => manager.auth().await,
+        None => None,
+    };
+    let auth = match auth {
         Some(auth) => auth,
         None => {
             eprintln!(
@@ -79,7 +82,7 @@ async fn init_backend(user_agent_suffix: &str) -> anyhow::Result<BackendContext>
         append_error_log(format!("auth: mode=ChatGPT account_id={acc}"));
     }
 
-    let token = match auth.get_token().await {
+    let token = match auth.get_token() {
         Ok(t) if !t.is_empty() => t,
         _ => {
             eprintln!(

codex-rs/cloud-tasks/src/util.rs

@@ -85,8 +85,8 @@ pub async fn build_chatgpt_headers() -> HeaderMap {
         HeaderValue::from_str(&ua).unwrap_or(HeaderValue::from_static("codex-cli")),
     );
     if let Some(am) = load_auth_manager().await
-        && let Some(auth) = am.auth()
-        && let Ok(tok) = auth.get_token().await
+        && let Some(auth) = am.auth().await
+        && let Ok(tok) = auth.get_token()
         && !tok.is_empty()
     {
         let v = format!("Bearer {tok}");

codex-rs/codex-api/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "codex-api",
+    crate_name = "codex_api",
+)

codex-rs/codex-api/src/endpoint/chat.rs

@@ -10,6 +10,7 @@ use crate::provider::WireApi;
 use crate::sse::chat::spawn_chat_stream;
 use crate::telemetry::SseTelemetry;
 use codex_client::HttpTransport;
+use codex_client::RequestCompression;
 use codex_client::RequestTelemetry;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ReasoningItemContent;
@@ -80,7 +81,13 @@ impl<T: HttpTransport, A: AuthProvider> ChatClient<T, A> {
         extra_headers: HeaderMap,
     ) -> Result<ResponseStream, ApiError> {
         self.streaming
-            .stream(self.path(), body, extra_headers, spawn_chat_stream)
+            .stream(
+                self.path(),
+                body,
+                extra_headers,
+                RequestCompression::None,
+                spawn_chat_stream,
+            )
             .await
     }
 }

codex-rs/codex-api/src/endpoint/responses.rs

@@ -9,9 +9,11 @@ use crate::provider::Provider;
 use crate::provider::WireApi;
 use crate::requests::ResponsesRequest;
 use crate::requests::ResponsesRequestBuilder;
+use crate::requests::responses::Compression;
 use crate::sse::spawn_response_stream;
 use crate::telemetry::SseTelemetry;
 use codex_client::HttpTransport;
+use codex_client::RequestCompression;
 use codex_client::RequestTelemetry;
 use codex_protocol::protocol::SessionSource;
 use http::HeaderMap;
@@ -33,6 +35,7 @@ pub struct ResponsesOptions {
     pub conversation_id: Option<String>,
     pub session_source: Option<SessionSource>,
     pub extra_headers: HeaderMap,
+    pub compression: Compression,
 }
 
 impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
@@ -56,7 +59,8 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
         &self,
         request: ResponsesRequest,
     ) -> Result<ResponseStream, ApiError> {
-        self.stream(request.body, request.headers).await
+        self.stream(request.body, request.headers, request.compression)
+            .await
     }
 
     #[instrument(level = "trace", skip_all, err)]
@@ -75,6 +79,7 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
             conversation_id,
             session_source,
             extra_headers,
+            compression,
         } = options;
 
         let request = ResponsesRequestBuilder::new(model, &prompt.instructions, &prompt.input)
@@ -88,6 +93,7 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
             .session_source(session_source)
             .store_override(store_override)
             .extra_headers(extra_headers)
+            .compression(compression)
             .build(self.streaming.provider())?;
 
         self.stream_request(request).await
@@ -104,9 +110,21 @@ impl<T: HttpTransport, A: AuthProvider> ResponsesClient<T, A> {
         &self,
         body: Value,
         extra_headers: HeaderMap,
+        compression: Compression,
     ) -> Result<ResponseStream, ApiError> {
+        let compression = match compression {
+            Compression::None => RequestCompression::None,
+            Compression::Zstd => RequestCompression::Zstd,
+        };
+
         self.streaming
-            .stream(self.path(), body, extra_headers, spawn_response_stream)
+            .stream(
+                self.path(),
+                body,
+                extra_headers,
+                compression,
+                spawn_response_stream,
+            )
             .await
     }
 }

codex-rs/codex-api/src/endpoint/streaming.rs

@@ -6,6 +6,7 @@ use crate::provider::Provider;
 use crate::telemetry::SseTelemetry;
 use crate::telemetry::run_with_request_telemetry;
 use codex_client::HttpTransport;
+use codex_client::RequestCompression;
 use codex_client::RequestTelemetry;
 use codex_client::StreamResponse;
 use http::HeaderMap;
@@ -52,6 +53,7 @@ impl<T: HttpTransport, A: AuthProvider> StreamingClient<T, A> {
         path: &str,
         body: Value,
         extra_headers: HeaderMap,
+        compression: RequestCompression,
         spawner: fn(StreamResponse, Duration, Option<Arc<dyn SseTelemetry>>) -> ResponseStream,
     ) -> Result<ResponseStream, ApiError> {
         let builder = || {
@@ -62,6 +64,7 @@ impl<T: HttpTransport, A: AuthProvider> StreamingClient<T, A> {
                 http::HeaderValue::from_static("text/event-stream"),
             );
             req.body = Some(body.clone());
+            req.compression = compression;
             add_auth_headers(&self.auth, req)
         };
 

codex-rs/codex-api/src/provider.rs

@@ -1,4 +1,5 @@
 use codex_client::Request;
+use codex_client::RequestCompression;
 use codex_client::RetryOn;
 use codex_client::RetryPolicy;
 use http::Method;
@@ -87,6 +88,7 @@ impl Provider {
             url: self.url_for_path(path),
             headers: self.headers.clone(),
             body: None,
+            compression: RequestCompression::None,
             timeout: None,
         }
     }

codex-rs/codex-api/src/requests/responses.rs

@@ -11,10 +11,18 @@ use codex_protocol::protocol::SessionSource;
 use http::HeaderMap;
 use serde_json::Value;
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum Compression {
+    #[default]
+    None,
+    Zstd,
+}
+
 /// Assembled request body plus headers for a Responses stream request.
 pub struct ResponsesRequest {
     pub body: Value,
     pub headers: HeaderMap,
+    pub compression: Compression,
 }
 
 #[derive(Default)]
@@ -32,6 +40,7 @@ pub struct ResponsesRequestBuilder<'a> {
     session_source: Option<SessionSource>,
     store_override: Option<bool>,
     headers: HeaderMap,
+    compression: Compression,
 }
 
 impl<'a> ResponsesRequestBuilder<'a> {
@@ -94,6 +103,11 @@ impl<'a> ResponsesRequestBuilder<'a> {
         self
     }
 
+    pub fn compression(mut self, compression: Compression) -> Self {
+        self.compression = compression;
+        self
+    }
+
     pub fn build(self, provider: &Provider) -> Result<ResponsesRequest, ApiError> {
         let model = self
             .model
@@ -138,7 +152,11 @@ impl<'a> ResponsesRequestBuilder<'a> {
             insert_header(&mut headers, "x-openai-subagent", &subagent);
         }
 
-        Ok(ResponsesRequest { body, headers })
+        Ok(ResponsesRequest {
+            body,
+            headers,
+            compression: self.compression,
+        })
     }
 }
 

codex-rs/codex-api/src/sse/responses.rs

@@ -301,7 +301,9 @@ pub async fn process_sse(
                     }
                 }
             }
-            _ => {}
+            _ => {
+                trace!("unhandled SSE event: {:#?}", event.kind);
+            }
         }
     }
 }

codex-rs/codex-api/tests/clients.rs

@@ -11,6 +11,7 @@ use codex_api::Provider;
 use codex_api::ResponsesClient;
 use codex_api::ResponsesOptions;
 use codex_api::WireApi;
+use codex_api::requests::responses::Compression;
 use codex_client::HttpTransport;
 use codex_client::Request;
 use codex_client::Response;
@@ -229,7 +230,9 @@ async fn responses_client_uses_responses_path_for_responses_wire() -> Result<()>
     let client = ResponsesClient::new(transport, provider("openai", WireApi::Responses), NoAuth);
 
     let body = serde_json::json!({ "echo": true });
-    let _stream = client.stream(body, HeaderMap::new()).await?;
+    let _stream = client
+        .stream(body, HeaderMap::new(), Compression::None)
+        .await?;
 
     let requests = state.take_stream_requests();
     assert_path_ends_with(&requests, "/responses");
@@ -243,7 +246,9 @@ async fn responses_client_uses_chat_path_for_chat_wire() -> Result<()> {
     let client = ResponsesClient::new(transport, provider("openai", WireApi::Chat), NoAuth);
 
     let body = serde_json::json!({ "echo": true });
-    let _stream = client.stream(body, HeaderMap::new()).await?;
+    let _stream = client
+        .stream(body, HeaderMap::new(), Compression::None)
+        .await?;
 
     let requests = state.take_stream_requests();
     assert_path_ends_with(&requests, "/chat/completions");
@@ -258,7 +263,9 @@ async fn streaming_client_adds_auth_headers() -> Result<()> {
     let client = ResponsesClient::new(transport, provider("openai", WireApi::Responses), auth);
 
     let body = serde_json::json!({ "model": "gpt-test" });
-    let _stream = client.stream(body, HeaderMap::new()).await?;
+    let _stream = client
+        .stream(body, HeaderMap::new(), Compression::None)
+        .await?;
 
     let requests = state.take_stream_requests();
     assert_eq!(requests.len(), 1);

codex-rs/codex-api/tests/sse_end_to_end.rs

@@ -9,6 +9,7 @@ use codex_api::Provider;
 use codex_api::ResponseEvent;
 use codex_api::ResponsesClient;
 use codex_api::WireApi;
+use codex_api::requests::responses::Compression;
 use codex_client::HttpTransport;
 use codex_client::Request;
 use codex_client::Response;
@@ -124,7 +125,11 @@ async fn responses_stream_parses_items_and_completed_end_to_end() -> Result<()>
     let client = ResponsesClient::new(transport, provider("openai", WireApi::Responses), NoAuth);
 
     let mut stream = client
-        .stream(serde_json::json!({"echo": true}), HeaderMap::new())
+        .stream(
+            serde_json::json!({"echo": true}),
+            HeaderMap::new(),
+            Compression::None,
+        )
         .await?;
 
     let mut events = Vec::new();
@@ -189,7 +194,11 @@ async fn responses_stream_aggregates_output_text_deltas() -> Result<()> {
     let client = ResponsesClient::new(transport, provider("openai", WireApi::Responses), NoAuth);
 
     let stream = client
-        .stream(serde_json::json!({"echo": true}), HeaderMap::new())
+        .stream(
+            serde_json::json!({"echo": true}),
+            HeaderMap::new(),
+            Compression::None,
+        )
         .await?;
 
     let mut stream = stream.aggregate();

codex-rs/codex-backend-openapi-models/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "codex-backend-openapi-models",
+    crate_name = "codex_backend_openapi_models",
+)

codex-rs/codex-client/BUILD.bazel

@@ -0,0 +1,6 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "codex-client",
+    crate_name = "codex_client",
+)

codex-rs/codex-client/Cargo.toml

@@ -19,6 +19,7 @@ thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt", "time", "sync"] }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
+zstd = { workspace = true }
 
 [lints]
 workspace = true

codex-rs/codex-client/src/default_client.rs

@@ -104,6 +104,13 @@ impl CodexRequestBuilder {
         self.map(|builder| builder.json(value))
     }
 
+    pub fn body<B>(self, body: B) -> Self
+    where
+        B: Into<reqwest::Body>,
+    {
+        self.map(|builder| builder.body(body))
+    }
+
     pub async fn send(self) -> Result<Response, reqwest::Error> {
         let headers = trace_headers();
 

codex-rs/codex-client/src/lib.rs

@@ -11,6 +11,7 @@ pub use crate::default_client::CodexRequestBuilder;
 pub use crate::error::StreamError;
 pub use crate::error::TransportError;
 pub use crate::request::Request;
+pub use crate::request::RequestCompression;
 pub use crate::request::Response;
 pub use crate::retry::RetryOn;
 pub use crate::retry::RetryPolicy;

codex-rs/codex-client/src/request.rs

@@ -5,12 +5,20 @@ use serde::Serialize;
 use serde_json::Value;
 use std::time::Duration;
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum RequestCompression {
+    #[default]
+    None,
+    Zstd,
+}
+
 #[derive(Debug, Clone)]
 pub struct Request {
     pub method: Method,
     pub url: String,
     pub headers: HeaderMap,
     pub body: Option<Value>,
+    pub compression: RequestCompression,
     pub timeout: Option<Duration>,
 }
 
@@ -21,6 +29,7 @@ impl Request {
             url,
             headers: HeaderMap::new(),
             body: None,
+            compression: RequestCompression::None,
             timeout: None,
         }
     }
@@ -29,6 +38,11 @@ impl Request {
         self.body = serde_json::to_value(body).ok();
         self
     }
+
+    pub fn with_compression(mut self, compression: RequestCompression) -> Self {
+        self.compression = compression;
+        self
+    }
 }
 
 #[derive(Debug, Clone)]

codex-rs/codex-client/src/transport.rs

@@ -2,6 +2,7 @@ use crate::default_client::CodexHttpClient;
 use crate::default_client::CodexRequestBuilder;
 use crate::error::TransportError;
 use crate::request::Request;
+use crate::request::RequestCompression;
 use crate::request::Response;
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -41,18 +42,70 @@ impl ReqwestTransport {
     }
 
     fn build(&self, req: Request) -> Result<CodexRequestBuilder, TransportError> {
-        let mut builder = self
-            .client
-            .request(
-                Method::from_bytes(req.method.as_str().as_bytes()).unwrap_or(Method::GET),
-                &req.url,
-            )
-            .headers(req.headers);
-        if let Some(timeout) = req.timeout {
+        let Request {
+            method,
+            url,
+            mut headers,
+            body,
+            compression,
+            timeout,
+        } = req;
+
+        let mut builder = self.client.request(
+            Method::from_bytes(method.as_str().as_bytes()).unwrap_or(Method::GET),
+            &url,
+        );
+
+        if let Some(timeout) = timeout {
             builder = builder.timeout(timeout);
         }
-        if let Some(body) = req.body {
-            builder = builder.json(&body);
+
+        if let Some(body) = body {
+            if compression != RequestCompression::None {
+                if headers.contains_key(http::header::CONTENT_ENCODING) {
+                    return Err(TransportError::Build(
+                        "request compression was requested but content-encoding is already set"
+                            .to_string(),
+                    ));
+                }
+
+                let json = serde_json::to_vec(&body)
+                    .map_err(|err| TransportError::Build(err.to_string()))?;
+                let pre_compression_bytes = json.len();
+                let compression_start = std::time::Instant::now();
+                let (compressed, content_encoding) = match compression {
+                    RequestCompression::None => unreachable!("guarded by compression != None"),
+                    RequestCompression::Zstd => (
+                        zstd::stream::encode_all(std::io::Cursor::new(json), 3)
+                            .map_err(|err| TransportError::Build(err.to_string()))?,
+                        http::HeaderValue::from_static("zstd"),
+                    ),
+                };
+                let post_compression_bytes = compressed.len();
+                let compression_duration = compression_start.elapsed();
+
+                // Ensure the server knows to unpack the request body.
+                headers.insert(http::header::CONTENT_ENCODING, content_encoding);
+                if !headers.contains_key(http::header::CONTENT_TYPE) {
+                    headers.insert(
+                        http::header::CONTENT_TYPE,
+                        http::HeaderValue::from_static("application/json"),
+                    );
+                }
+
+                tracing::info!(
+                    pre_compression_bytes,
+                    post_compression_bytes,
+                    compression_duration_ms = compression_duration.as_millis(),
+                    "Compressed request body with zstd"
+                );
+
+                builder = builder.headers(headers).body(compressed);
+            } else {
+                builder = builder.headers(headers).json(&body);
+            }
+        } else {
+            builder = builder.headers(headers);
         }
         Ok(builder)
     }

codex-rs/common/BUILD.bazel

@@ -0,0 +1,11 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "common",
+    crate_name = "codex_common",
+    crate_features = [
+        "cli",
+        "elapsed",
+        "sandbox_summary",
+    ],
+)

codex-rs/core/BUILD.bazel

@@ -0,0 +1,31 @@
+load("//:defs.bzl", "codex_rust_crate")
+
+codex_rust_crate(
+    name = "core",
+    crate_name = "codex_core",
+    # TODO(mbolin): Eliminate the use of features in the version of the
+    # rust_library() that is used by rust_binary() rules for release artifacts
+    # such as the Codex CLI.
+    crate_features = ["deterministic_process_ids", "test-support"],
+    compile_data = glob(
+        include = ["**"],
+        exclude = [
+            "**/* *",
+            "BUILD.bazel",
+            "Cargo.toml",
+        ],
+        allow_empty = True,
+    ),
+    integration_compile_data_extra = [
+        "//codex-rs/apply-patch:apply_patch_tool_instructions.md",
+        "prompt.md",
+    ],
+    integration_deps_extra = ["//codex-rs/core/tests/common:common"],
+    test_tags = ["no-sandbox"],
+    extra_binaries = [
+        "//codex-rs/linux-sandbox:codex-linux-sandbox",
+        "//codex-rs/rmcp-client:test_stdio_server",
+        "//codex-rs/rmcp-client:test_streamable_http_server",
+        "//codex-rs/cli:codex",
+    ],
+)

codex-rs/core/Cargo.toml

@@ -122,11 +122,11 @@ keyring = { workspace = true, features = ["sync-secret-service"] }
 assert_cmd = { workspace = true }
 assert_matches = { workspace = true }
 codex-arg0 = { workspace = true }
-codex-core = { path = ".", features = ["deterministic_process_ids"] }
+codex-core = { path = ".", default-features = false, features = ["deterministic_process_ids"] }
+codex-otel = { workspace = true, features = ["disable-default-metrics-exporter"] }
 codex-utils-cargo-bin = { workspace = true }
 core_test_support = { workspace = true }
 ctor = { workspace = true }
-escargot = { workspace = true }
 image = { workspace = true, features = ["jpeg", "png"] }
 maplit = { workspace = true }
 predicates = { workspace = true }
@@ -137,6 +137,7 @@ tracing-subscriber = { workspace = true }
 tracing-test = { workspace = true, features = ["no-env-filter"] }
 walkdir = { workspace = true }
 wiremock = { workspace = true }
+zstd = { workspace = true }
 
 [package.metadata.cargo-shear]
 ignored = ["openssl-sys"]

codex-rs/core/src/agent/control.rs

@@ -116,10 +116,10 @@ mod tests {
     use super::*;
     use crate::agent::agent_status_from_event;
     use codex_protocol::protocol::ErrorEvent;
-    use codex_protocol::protocol::TaskCompleteEvent;
-    use codex_protocol::protocol::TaskStartedEvent;
     use codex_protocol::protocol::TurnAbortReason;
     use codex_protocol::protocol::TurnAbortedEvent;
+    use codex_protocol::protocol::TurnCompleteEvent;
+    use codex_protocol::protocol::TurnStartedEvent;
     use pretty_assertions::assert_eq;
 
     #[tokio::test]
@@ -144,7 +144,7 @@ mod tests {
 
     #[tokio::test]
     async fn on_event_updates_status_from_task_started() {
-        let status = agent_status_from_event(&EventMsg::TaskStarted(TaskStartedEvent {
+        let status = agent_status_from_event(&EventMsg::TurnStarted(TurnStartedEvent {
             model_context_window: None,
         }));
         assert_eq!(status, Some(AgentStatus::Running));
@@ -152,7 +152,7 @@ mod tests {
 
     #[tokio::test]
     async fn on_event_updates_status_from_task_complete() {
-        let status = agent_status_from_event(&EventMsg::TaskComplete(TaskCompleteEvent {
+        let status = agent_status_from_event(&EventMsg::TurnComplete(TurnCompleteEvent {
             last_agent_message: Some("done".to_string()),
         }));
         let expected = AgentStatus::Completed(Some("done".to_string()));

codex-rs/core/src/agent/status.rs

@@ -5,8 +5,8 @@ use codex_protocol::protocol::EventMsg;
 /// Returns `None` when the event does not affect status tracking.
 pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option<AgentStatus> {
     match msg {
-        EventMsg::TaskStarted(_) => Some(AgentStatus::Running),
-        EventMsg::TaskComplete(ev) => Some(AgentStatus::Completed(ev.last_agent_message.clone())),
+        EventMsg::TurnStarted(_) => Some(AgentStatus::Running),
+        EventMsg::TurnComplete(ev) => Some(AgentStatus::Completed(ev.last_agent_message.clone())),
         EventMsg::TurnAborted(ev) => Some(AgentStatus::Errored(format!("{:?}", ev.reason))),
         EventMsg::Error(ev) => Some(AgentStatus::Errored(ev.message.clone())),
         EventMsg::ShutdownComplete => Some(AgentStatus::Shutdown),

codex-rs/core/src/api_bridge.rs

@@ -100,7 +100,7 @@ fn extract_request_id(headers: Option<&HeaderMap>) -> Option<String> {
     })
 }
 
-pub(crate) async fn auth_provider_from_auth(
+pub(crate) fn auth_provider_from_auth(
     auth: Option<CodexAuth>,
     provider: &ModelProviderInfo,
 ) -> crate::error::Result<CoreAuthProvider> {
@@ -119,7 +119,7 @@ pub(crate) async fn auth_provider_from_auth(
     }
 
     if let Some(auth) = auth {
-        let token = auth.get_token().await?;
+        let token = auth.get_token()?;
         Ok(CoreAuthProvider {
             token: Some(token),
             account_id: auth.get_account_id(),

codex-rs/core/src/auth.rs

@@ -8,12 +8,10 @@ use serde::Serialize;
 use serial_test::serial;
 use std::env;
 use std::fmt::Debug;
-use std::io::ErrorKind;
 use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::time::Duration;
 
 use codex_app_server_protocol::AuthMode;
 use codex_protocol::config_types::ForcedLoginMethod;
@@ -77,10 +75,6 @@ impl RefreshTokenError {
             Self::Transient(_) => None,
         }
     }
-
-    fn other_with_message(message: impl Into<String>) -> Self {
-        Self::Transient(std::io::Error::other(message.into()))
-    }
 }
 
 impl From<RefreshTokenError> for std::io::Error {
@@ -93,40 +87,6 @@ impl From<RefreshTokenError> for std::io::Error {
 }
 
 impl CodexAuth {
-    pub async fn refresh_token(&self) -> Result<String, RefreshTokenError> {
-        tracing::info!("Refreshing token");
-
-        let token_data = self.get_current_token_data().ok_or_else(|| {
-            RefreshTokenError::Transient(std::io::Error::other("Token data is not available."))
-        })?;
-        let token = token_data.refresh_token;
-
-        let refresh_response = try_refresh_token(token, &self.client).await?;
-
-        let updated = update_tokens(
-            &self.storage,
-            refresh_response.id_token,
-            refresh_response.access_token,
-            refresh_response.refresh_token,
-        )
-        .await
-        .map_err(RefreshTokenError::from)?;
-
-        if let Ok(mut auth_lock) = self.auth_dot_json.lock() {
-            *auth_lock = Some(updated.clone());
-        }
-
-        let access = match updated.tokens {
-            Some(t) => t.access_token,
-            None => {
-                return Err(RefreshTokenError::other_with_message(
-                    "Token data is not available after refresh.",
-                ));
-            }
-        };
-        Ok(access)
-    }
-
     /// Loads the available auth information from auth storage.
     pub fn from_auth_storage(
         codex_home: &Path,
@@ -135,62 +95,23 @@ impl CodexAuth {
         load_auth(codex_home, false, auth_credentials_store_mode)
     }
 
-    pub async fn get_token_data(&self) -> Result<TokenData, std::io::Error> {
+    pub fn get_token_data(&self) -> Result<TokenData, std::io::Error> {
         let auth_dot_json: Option<AuthDotJson> = self.get_current_auth_json();
         match auth_dot_json {
             Some(AuthDotJson {
-                tokens: Some(mut tokens),
-                last_refresh: Some(last_refresh),
+                tokens: Some(tokens),
+                last_refresh: Some(_),
                 ..
-            }) => {
-                if last_refresh < Utc::now() - chrono::Duration::days(TOKEN_REFRESH_INTERVAL) {
-                    let refresh_result = tokio::time::timeout(
-                        Duration::from_secs(60),
-                        try_refresh_token(tokens.refresh_token.clone(), &self.client),
-                    )
-                    .await;
-                    let refresh_response = match refresh_result {
-                        Ok(Ok(response)) => response,
-                        Ok(Err(err)) => return Err(err.into()),
-                        Err(_) => {
-                            return Err(std::io::Error::new(
-                                ErrorKind::TimedOut,
-                                "timed out while refreshing OpenAI API key",
-                            ));
-                        }
-                    };
-
-                    let updated_auth_dot_json = update_tokens(
-                        &self.storage,
-                        refresh_response.id_token,
-                        refresh_response.access_token,
-                        refresh_response.refresh_token,
-                    )
-                    .await?;
-
-                    tokens = updated_auth_dot_json
-                        .tokens
-                        .clone()
-                        .ok_or(std::io::Error::other(
-                            "Token data is not available after refresh.",
-                        ))?;
-
-                    #[expect(clippy::unwrap_used)]
-                    let mut auth_lock = self.auth_dot_json.lock().unwrap();
-                    *auth_lock = Some(updated_auth_dot_json);
-                }
-
-                Ok(tokens)
-            }
+            }) => Ok(tokens),
             _ => Err(std::io::Error::other("Token data is not available.")),
         }
     }
 
-    pub async fn get_token(&self) -> Result<String, std::io::Error> {
+    pub fn get_token(&self) -> Result<String, std::io::Error> {
         match self.mode {
             AuthMode::ApiKey => Ok(self.api_key.clone().unwrap_or_default()),
             AuthMode::ChatGPT => {
-                let id_token = self.get_token_data().await?.access_token;
+                let id_token = self.get_token_data()?.access_token;
                 Ok(id_token)
             }
         }
@@ -338,7 +259,7 @@ pub fn load_auth_dot_json(
     storage.load()
 }
 
-pub async fn enforce_login_restrictions(config: &Config) -> std::io::Result<()> {
+pub fn enforce_login_restrictions(config: &Config) -> std::io::Result<()> {
     let Some(auth) = load_auth(
         &config.codex_home,
         true,
@@ -376,7 +297,7 @@ pub async fn enforce_login_restrictions(config: &Config) -> std::io::Result<()>
             return Ok(());
         }
 
-        let token_data = match auth.get_token_data().await {
+        let token_data = match auth.get_token_data() {
             Ok(data) => data,
             Err(err) => {
                 return logout_with_message(
@@ -525,6 +446,7 @@ async fn try_refresh_token(
         Ok(refresh_response)
     } else {
         let body = response.text().await.unwrap_or_default();
+        tracing::error!("Failed to refresh token: {status}: {body}");
         if status == StatusCode::UNAUTHORIZED {
             let failed = classify_refresh_token_failure(&body);
             Err(RefreshTokenError::Permanent(failed))
@@ -623,6 +545,89 @@ struct CachedAuth {
     auth: Option<CodexAuth>,
 }
 
+enum UnauthorizedRecoveryStep {
+    Reload,
+    RefreshToken,
+    Done,
+}
+
+enum ReloadOutcome {
+    Reloaded,
+    Skipped,
+}
+
+// UnauthorizedRecovery is a state machine that handles an attempt to refresh the authentication when requests
+// to API fail with 401 status code.
+// The client calls next() every time it encounters a 401 error, one time per retry.
+// For API key based authentication, we don't do anything and let the error bubble to the user.
+// For ChatGPT based authentication, we:
+// 1. Attempt to reload the auth data from disk. We only reload if the account id matches the one the current process is running as.
+// 2. Attempt to refresh the token using OAuth token refresh flow.
+// If after both steps the server still responds with 401 we let the error bubble to the user.
+pub struct UnauthorizedRecovery {
+    manager: Arc<AuthManager>,
+    step: UnauthorizedRecoveryStep,
+    expected_account_id: Option<String>,
+}
+
+impl UnauthorizedRecovery {
+    fn new(manager: Arc<AuthManager>) -> Self {
+        let expected_account_id = manager
+            .auth_cached()
+            .as_ref()
+            .and_then(CodexAuth::get_account_id);
+        Self {
+            manager,
+            step: UnauthorizedRecoveryStep::Reload,
+            expected_account_id,
+        }
+    }
+
+    pub fn has_next(&self) -> bool {
+        if !self
+            .manager
+            .auth_cached()
+            .is_some_and(|auth| auth.mode == AuthMode::ChatGPT)
+        {
+            return false;
+        }
+
+        !matches!(self.step, UnauthorizedRecoveryStep::Done)
+    }
+
+    pub async fn next(&mut self) -> Result<(), RefreshTokenError> {
+        if !self.has_next() {
+            return Err(RefreshTokenError::Permanent(RefreshTokenFailedError::new(
+                RefreshTokenFailedReason::Other,
+                "No more recovery steps available.",
+            )));
+        }
+
+        match self.step {
+            UnauthorizedRecoveryStep::Reload => {
+                match self
+                    .manager
+                    .reload_if_account_id_matches(self.expected_account_id.as_deref())
+                {
+                    ReloadOutcome::Reloaded => {
+                        self.step = UnauthorizedRecoveryStep::RefreshToken;
+                    }
+                    ReloadOutcome::Skipped => {
+                        self.manager.refresh_token().await?;
+                        self.step = UnauthorizedRecoveryStep::Done;
+                    }
+                }
+            }
+            UnauthorizedRecoveryStep::RefreshToken => {
+                self.manager.refresh_token().await?;
+                self.step = UnauthorizedRecoveryStep::Done;
+            }
+            UnauthorizedRecoveryStep::Done => {}
+        }
+        Ok(())
+    }
+}
+
 /// Central manager providing a single source of truth for auth.json derived
 /// authentication data. It loads once (or on preference change) and then
 /// hands out cloned `CodexAuth` values so the rest of the program has a
@@ -689,28 +694,53 @@ impl AuthManager {
         })
     }
 
-    /// Current cached auth (clone). May be `None` if not logged in or load failed.
-    pub fn auth(&self) -> Option<CodexAuth> {
+    /// Current cached auth (clone) without attempting a refresh.
+    pub fn auth_cached(&self) -> Option<CodexAuth> {
         self.inner.read().ok().and_then(|c| c.auth.clone())
     }
 
+    /// Current cached auth (clone). May be `None` if not logged in or load failed.
+    /// Refreshes cached ChatGPT tokens if they are stale before returning.
+    pub async fn auth(&self) -> Option<CodexAuth> {
+        let auth = self.auth_cached()?;
+        if let Err(err) = self.refresh_if_stale(&auth).await {
+            tracing::error!("Failed to refresh token: {}", err);
+            return Some(auth);
+        }
+        self.auth_cached()
+    }
+
     /// Force a reload of the auth information from auth.json. Returns
     /// whether the auth value changed.
     pub fn reload(&self) -> bool {
-        let new_auth = load_auth(
-            &self.codex_home,
-            self.enable_codex_api_key_env,
-            self.auth_credentials_store_mode,
-        )
-        .ok()
-        .flatten();
-        if let Ok(mut guard) = self.inner.write() {
-            let changed = !AuthManager::auths_equal(&guard.auth, &new_auth);
-            guard.auth = new_auth;
-            changed
-        } else {
-            false
+        tracing::info!("Reloading auth");
+        let new_auth = self.load_auth_from_storage();
+        self.set_auth(new_auth)
+    }
+
+    fn reload_if_account_id_matches(&self, expected_account_id: Option<&str>) -> ReloadOutcome {
+        let expected_account_id = match expected_account_id {
+            Some(account_id) => account_id,
+            None => {
+                tracing::info!("Skipping auth reload because no account id is available.");
+                return ReloadOutcome::Skipped;
+            }
+        };
+
+        let new_auth = self.load_auth_from_storage();
+        let new_account_id = new_auth.as_ref().and_then(CodexAuth::get_account_id);
+
+        if new_account_id.as_deref() != Some(expected_account_id) {
+            let found_account_id = new_account_id.as_deref().unwrap_or("unknown");
+            tracing::info!(
+                "Skipping auth reload due to account id mismatch (expected: {expected_account_id}, found: {found_account_id})"
+            );
+            return ReloadOutcome::Skipped;
         }
+
+        tracing::info!("Reloading auth for account {expected_account_id}");
+        self.set_auth(new_auth);
+        ReloadOutcome::Reloaded
     }
 
     fn auths_equal(a: &Option<CodexAuth>, b: &Option<CodexAuth>) -> bool {
@@ -721,6 +751,27 @@ impl AuthManager {
         }
     }
 
+    fn load_auth_from_storage(&self) -> Option<CodexAuth> {
+        load_auth(
+            &self.codex_home,
+            self.enable_codex_api_key_env,
+            self.auth_credentials_store_mode,
+        )
+        .ok()
+        .flatten()
+    }
+
+    fn set_auth(&self, new_auth: Option<CodexAuth>) -> bool {
+        if let Ok(mut guard) = self.inner.write() {
+            let changed = !AuthManager::auths_equal(&guard.auth, &new_auth);
+            tracing::info!("Reloaded auth, changed: {changed}");
+            guard.auth = new_auth;
+            changed
+        } else {
+            false
+        }
+    }
+
     /// Convenience constructor returning an `Arc` wrapper.
     pub fn shared(
         codex_home: PathBuf,
@@ -734,26 +785,27 @@ impl AuthManager {
         ))
     }
 
+    pub fn unauthorized_recovery(self: &Arc<Self>) -> UnauthorizedRecovery {
+        UnauthorizedRecovery::new(Arc::clone(self))
+    }
+
     /// Attempt to refresh the current auth token (if any). On success, reload
     /// the auth state from disk so other components observe refreshed token.
-    /// If the token refresh fails in a permanent (non‑transient) way, logs out
-    /// to clear invalid auth state.
-    pub async fn refresh_token(&self) -> Result<Option<String>, RefreshTokenError> {
-        let auth = match self.auth() {
-            Some(a) => a,
-            None => return Ok(None),
+    /// If the token refresh fails, returns the error to the caller.
+    pub async fn refresh_token(&self) -> Result<(), RefreshTokenError> {
+        tracing::info!("Refreshing token");
+
+        let auth = match self.auth_cached() {
+            Some(auth) => auth,
+            None => return Ok(()),
         };
-        match auth.refresh_token().await {
-            Ok(token) => {
-                // Reload to pick up persisted changes.
-                self.reload();
-                Ok(Some(token))
-            }
-            Err(e) => {
-                tracing::error!("Failed to refresh token: {}", e);
-                Err(e)
-            }
-        }
+        let token_data = auth.get_current_token_data().ok_or_else(|| {
+            RefreshTokenError::Transient(std::io::Error::other("Token data is not available."))
+        })?;
+        self.refresh_tokens(&auth, token_data.refresh_token).await?;
+        // Reload to pick up persisted changes.
+        self.reload();
+        Ok(())
     }
 
     /// Log out by deleting the on‑disk auth.json (if present). Returns Ok(true)
@@ -768,7 +820,51 @@ impl AuthManager {
     }
 
     pub fn get_auth_mode(&self) -> Option<AuthMode> {
-        self.auth().map(|a| a.mode)
+        self.auth_cached().map(|a| a.mode)
+    }
+
+    async fn refresh_if_stale(&self, auth: &CodexAuth) -> Result<bool, RefreshTokenError> {
+        if auth.mode != AuthMode::ChatGPT {
+            return Ok(false);
+        }
+
+        let auth_dot_json = match auth.get_current_auth_json() {
+            Some(auth_dot_json) => auth_dot_json,
+            None => return Ok(false),
+        };
+        let tokens = match auth_dot_json.tokens {
+            Some(tokens) => tokens,
+            None => return Ok(false),
+        };
+        let last_refresh = match auth_dot_json.last_refresh {
+            Some(last_refresh) => last_refresh,
+            None => return Ok(false),
+        };
+        if last_refresh >= Utc::now() - chrono::Duration::days(TOKEN_REFRESH_INTERVAL) {
+            return Ok(false);
+        }
+        self.refresh_tokens(auth, tokens.refresh_token).await?;
+        self.reload();
+        Ok(true)
+    }
+
+    async fn refresh_tokens(
+        &self,
+        auth: &CodexAuth,
+        refresh_token: String,
+    ) -> Result<(), RefreshTokenError> {
+        let refresh_response = try_refresh_token(refresh_token, &auth.client).await?;
+
+        update_tokens(
+            &auth.storage,
+            refresh_response.id_token,
+            refresh_response.access_token,
+            refresh_response.refresh_token,
+        )
+        .await
+        .map_err(RefreshTokenError::from)?;
+
+        Ok(())
     }
 }
 
@@ -930,7 +1026,7 @@ mod tests {
         assert_eq!(auth.mode, AuthMode::ApiKey);
         assert_eq!(auth.api_key, Some("sk-test-key".to_string()));
 
-        assert!(auth.get_token_data().await.is_err());
+        assert!(auth.get_token_data().is_err());
     }
 
     #[test]
@@ -1058,7 +1154,6 @@ mod tests {
         let config = build_config(codex_home.path(), Some(ForcedLoginMethod::Chatgpt), None).await;
 
         let err = super::enforce_login_restrictions(&config)
-            .await
             .expect_err("expected method mismatch to error");
         assert!(err.to_string().contains("ChatGPT login is required"));
         assert!(
@@ -1084,7 +1179,6 @@ mod tests {
         let config = build_config(codex_home.path(), None, Some("org_mine".to_string())).await;
 
         let err = super::enforce_login_restrictions(&config)
-            .await
             .expect_err("expected workspace mismatch to error");
         assert!(err.to_string().contains("workspace org_mine"));
         assert!(
@@ -1109,9 +1203,7 @@ mod tests {
 
         let config = build_config(codex_home.path(), None, Some("org_mine".to_string())).await;
 
-        super::enforce_login_restrictions(&config)
-            .await
-            .expect("matching workspace should succeed");
+        super::enforce_login_restrictions(&config).expect("matching workspace should succeed");
         assert!(
             codex_home.path().join("auth.json").exists(),
             "auth.json should remain when restrictions pass"
@@ -1127,9 +1219,7 @@ mod tests {
 
         let config = build_config(codex_home.path(), None, Some("org_mine".to_string())).await;
 
-        super::enforce_login_restrictions(&config)
-            .await
-            .expect("matching workspace should succeed");
+        super::enforce_login_restrictions(&config).expect("matching workspace should succeed");
         assert!(
             codex_home.path().join("auth.json").exists(),
             "auth.json should remain when restrictions pass"
@@ -1145,7 +1235,6 @@ mod tests {
         let config = build_config(codex_home.path(), Some(ForcedLoginMethod::Chatgpt), None).await;
 
         let err = super::enforce_login_restrictions(&config)
-            .await
             .expect_err("environment API key should not satisfy forced ChatGPT login");
         assert!(
             err.to_string()

codex-rs/core/src/client.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use crate::api_bridge::auth_provider_from_auth;
 use crate::api_bridge::map_api_error;
+use crate::auth::UnauthorizedRecovery;
 use codex_api::AggregateStreamExt;
 use codex_api::ChatClient as ApiChatClient;
 use codex_api::CompactClient as ApiCompactClient;
@@ -17,8 +18,10 @@ use codex_api::TransportError;
 use codex_api::common::Reasoning;
 use codex_api::create_text_param_for_request;
 use codex_api::error::ApiError;
+use codex_api::requests::responses::Compression;
 use codex_app_server_protocol::AuthMode;
-use codex_otel::otel_manager::OtelManager;
+use codex_otel::OtelManager;
+
 use codex_protocol::ThreadId;
 use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
 use codex_protocol::models::ResponseItem;
@@ -47,6 +50,7 @@ use crate::default_client::build_reqwest_client;
 use crate::error::CodexErr;
 use crate::error::Result;
 use crate::features::FEATURES;
+use crate::features::Feature;
 use crate::flags::CODEX_RS_SSE_FIXTURE;
 use crate::model_provider_info::ModelProviderInfo;
 use crate::model_provider_info::WireApi;
@@ -153,13 +157,18 @@ impl ModelClient {
         let conversation_id = self.conversation_id.to_string();
         let session_source = self.session_source.clone();
 
-        let mut refreshed = false;
+        let mut auth_recovery = auth_manager
+            .as_ref()
+            .map(super::auth::AuthManager::unauthorized_recovery);
         loop {
-            let auth = auth_manager.as_ref().and_then(|m| m.auth());
+            let auth = match auth_manager.as_ref() {
+                Some(manager) => manager.auth().await,
+                None => None,
+            };
             let api_provider = self
                 .provider
                 .to_api_provider(auth.as_ref().map(|a| a.mode))?;
-            let api_auth = auth_provider_from_auth(auth.clone(), &self.provider).await?;
+            let api_auth = auth_provider_from_auth(auth.clone(), &self.provider)?;
             let transport = ReqwestTransport::new(build_reqwest_client());
             let (request_telemetry, sse_telemetry) = self.build_streaming_telemetry();
             let client = ApiChatClient::new(transport, api_provider, api_auth)
@@ -179,7 +188,7 @@ impl ModelClient {
                 Err(ApiError::Transport(TransportError::Http { status, .. }))
                     if status == StatusCode::UNAUTHORIZED =>
                 {
-                    handle_unauthorized(status, &mut refreshed, &auth_manager, &auth).await?;
+                    handle_unauthorized(status, &mut auth_recovery).await?;
                     continue;
                 }
                 Err(err) => return Err(map_api_error(err)),
@@ -241,15 +250,34 @@ impl ModelClient {
         let conversation_id = self.conversation_id.to_string();
         let session_source = self.session_source.clone();
 
-        let mut refreshed = false;
+        let mut auth_recovery = auth_manager
+            .as_ref()
+            .map(super::auth::AuthManager::unauthorized_recovery);
         loop {
-            let auth = auth_manager.as_ref().and_then(|m| m.auth());
+            let auth = match auth_manager.as_ref() {
+                Some(manager) => manager.auth().await,
+                None => None,
+            };
             let api_provider = self
                 .provider
                 .to_api_provider(auth.as_ref().map(|a| a.mode))?;
-            let api_auth = auth_provider_from_auth(auth.clone(), &self.provider).await?;
+            let api_auth = auth_provider_from_auth(auth.clone(), &self.provider)?;
             let transport = ReqwestTransport::new(build_reqwest_client());
             let (request_telemetry, sse_telemetry) = self.build_streaming_telemetry();
+            let compression = if self
+                .config
+                .features
+                .enabled(Feature::EnableRequestCompression)
+                && auth
+                    .as_ref()
+                    .is_some_and(|auth| auth.mode == AuthMode::ChatGPT)
+                && self.provider.is_openai()
+            {
+                Compression::Zstd
+            } else {
+                Compression::None
+            };
+
             let client = ApiResponsesClient::new(transport, api_provider, api_auth)
                 .with_telemetry(Some(request_telemetry), Some(sse_telemetry));
 
@@ -262,6 +290,7 @@ impl ModelClient {
                 conversation_id: Some(conversation_id.clone()),
                 session_source: Some(session_source.clone()),
                 extra_headers: beta_feature_headers(&self.config),
+                compression,
             };
 
             let stream_result = client
@@ -275,7 +304,7 @@ impl ModelClient {
                 Err(ApiError::Transport(TransportError::Http { status, .. }))
                     if status == StatusCode::UNAUTHORIZED =>
                 {
-                    handle_unauthorized(status, &mut refreshed, &auth_manager, &auth).await?;
+                    handle_unauthorized(status, &mut auth_recovery).await?;
                     continue;
                 }
                 Err(err) => return Err(map_api_error(err)),
@@ -327,11 +356,14 @@ impl ModelClient {
             return Ok(Vec::new());
         }
         let auth_manager = self.auth_manager.clone();
-        let auth = auth_manager.as_ref().and_then(|m| m.auth());
+        let auth = match auth_manager.as_ref() {
+            Some(manager) => manager.auth().await,
+            None => None,
+        };
         let api_provider = self
             .provider
             .to_api_provider(auth.as_ref().map(|a| a.mode))?;
-        let api_auth = auth_provider_from_auth(auth.clone(), &self.provider).await?;
+        let api_auth = auth_provider_from_auth(auth.clone(), &self.provider)?;
         let transport = ReqwestTransport::new(build_reqwest_client());
         let request_telemetry = self.build_request_telemetry();
         let client = ApiCompactClient::new(transport, api_provider, api_auth)
@@ -483,29 +515,19 @@ where
 /// the mapped `CodexErr` is returned to the caller.
 async fn handle_unauthorized(
     status: StatusCode,
-    refreshed: &mut bool,
-    auth_manager: &Option<Arc<AuthManager>>,
-    auth: &Option<crate::auth::CodexAuth>,
+    auth_recovery: &mut Option<UnauthorizedRecovery>,
 ) -> Result<()> {
-    if *refreshed {
-        return Err(map_unauthorized_status(status));
-    }
-
-    if let Some(manager) = auth_manager.as_ref()
-        && let Some(auth) = auth.as_ref()
-        && auth.mode == AuthMode::ChatGPT
+    if let Some(recovery) = auth_recovery
+        && recovery.has_next()
     {
-        match manager.refresh_token().await {
-            Ok(_) => {
-                *refreshed = true;
-                Ok(())
-            }
+        return match recovery.next().await {
+            Ok(_) => Ok(()),
             Err(RefreshTokenError::Permanent(failed)) => Err(CodexErr::RefreshTokenFailed(failed)),
             Err(RefreshTokenError::Transient(other)) => Err(CodexErr::Io(other)),
-        }
-    } else {
-        Err(map_unauthorized_status(status))
+        };
     }
+
+    Err(map_unauthorized_status(status))
 }
 
 fn map_unauthorized_status(status: StatusCode) -> CodexErr {

codex-rs/core/src/codex.rs

@@ -7,6 +7,7 @@ use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 
 use crate::AuthManager;
+use crate::CodexAuth;
 use crate::SandboxState;
 use crate::agent::AgentControl;
 use crate::agent::AgentStatus;
@@ -43,9 +44,9 @@ use codex_protocol::protocol::RawResponseItemEvent;
 use codex_protocol::protocol::ReviewRequest;
 use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::SessionSource;
-use codex_protocol::protocol::TaskStartedEvent;
 use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::protocol::TurnContextItem;
+use codex_protocol::protocol::TurnStartedEvent;
 use codex_rmcp_client::ElicitationResponse;
 use futures::future::BoxFuture;
 use futures::prelude::*;
@@ -152,7 +153,7 @@ use crate::user_instructions::UserInstructions;
 use crate::user_notification::UserNotification;
 use crate::util::backoff;
 use codex_async_utils::OrCancelExt;
-use codex_otel::otel_manager::OtelManager;
+use codex_otel::OtelManager;
 use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseInputItem;
@@ -484,7 +485,7 @@ pub(crate) struct SessionSettingsUpdate {
 
 impl Session {
     /// Don't expand the number of mutated arguments on config. We are in the process of getting rid of it.
-    fn build_per_turn_config(session_configuration: &SessionConfiguration) -> Config {
+    pub(crate) fn build_per_turn_config(session_configuration: &SessionConfiguration) -> Config {
         // todo(aibrahim): store this state somewhere else so we don't need to mut config
         let config = session_configuration.original_config_do_not_use.clone();
         let mut per_turn_config = (*config).clone();
@@ -633,17 +634,32 @@ impl Session {
         }
         maybe_push_chat_wire_api_deprecation(&config, &mut post_session_configured_events);
 
+        let auth = auth_manager.auth().await;
+        let auth = auth.as_ref();
         let otel_manager = OtelManager::new(
             conversation_id,
             session_configuration.model.as_str(),
             session_configuration.model.as_str(),
-            auth_manager.auth().and_then(|a| a.get_account_id()),
-            auth_manager.auth().and_then(|a| a.get_account_email()),
-            auth_manager.auth().map(|a| a.mode),
+            auth.and_then(CodexAuth::get_account_id),
+            auth.and_then(CodexAuth::get_account_email),
+            auth.map(|a| a.mode),
             config.otel.log_user_prompt,
             terminal::user_agent(),
             session_configuration.session_source.clone(),
         );
+        config.features.emit_metrics(&otel_manager);
+        otel_manager.counter(
+            "codex.thread.started",
+            1,
+            &[(
+                "is_git",
+                if get_git_repo_root(&session_configuration.cwd).is_some() {
+                    "true"
+                } else {
+                    "false"
+                },
+            )],
+        );
 
         otel_manager.conversation_starts(
             config.model_provider.name.as_str(),
@@ -1243,12 +1259,10 @@ impl Session {
                     );
                 }
                 RolloutItem::Compacted(compacted) => {
-                    let snapshot = history.get_history();
-                    // TODO(jif) clean
                     if let Some(replacement) = &compacted.replacement_history {
                         history.replace(replacement.clone());
                     } else {
-                        let user_messages = collect_user_messages(&snapshot);
+                        let user_messages = collect_user_messages(history.raw_items());
                         let rebuilt = compact::build_compacted_history(
                             self.build_initial_context(turn_context),
                             &user_messages,
@@ -1263,7 +1277,7 @@ impl Session {
                 _ => {}
             }
         }
-        history.get_history()
+        history.raw_items().to_vec()
     }
 
     /// Append ResponseItems to the in-memory conversation history only.
@@ -1756,6 +1770,7 @@ mod handlers {
     use codex_protocol::protocol::TurnAbortReason;
     use codex_protocol::protocol::WarningEvent;
 
+    use crate::context_manager::is_user_turn_boundary;
     use codex_protocol::user_input::UserInput;
     use codex_rmcp_client::ElicitationAction;
     use codex_rmcp_client::ElicitationResponse;
@@ -2093,7 +2108,10 @@ mod handlers {
 
         let mut history = sess.clone_history().await;
         history.drop_last_n_user_turns(num_turns);
-        sess.replace_history(history.get_history()).await;
+
+        // Replace with the raw items. We don't want to replace with a normalized
+        // version of the history.
+        sess.replace_history(history.raw_items().to_vec()).await;
         sess.recompute_token_usage(turn_context.as_ref()).await;
 
         sess.send_event_raw_flushed(Event {
@@ -2110,6 +2128,17 @@ mod handlers {
             .terminate_all_processes()
             .await;
         info!("Shutting down Codex instance");
+        let history = sess.clone_history().await;
+        let turn_count = history
+            .raw_items()
+            .iter()
+            .filter(|item| is_user_turn_boundary(item))
+            .count();
+        sess.services.otel_manager.counter(
+            "conversation.turn.count",
+            i64::try_from(turn_count).unwrap_or(0),
+            &[],
+        );
 
         // Gracefully flush and shutdown rollout recorder on session end so tests
         // that inspect the rollout file do not race with the background writer.
@@ -2295,9 +2324,9 @@ fn errors_to_info(errors: &[SkillError]) -> Vec<SkillErrorInfo> {
 /// - If the model requests a function call, we execute it and send the output
 ///   back to the model in the next turn.
 /// - If the model sends only an assistant message, we record it in the
-///   conversation history and consider the task complete.
+///   conversation history and consider the turn complete.
 ///
-pub(crate) async fn run_task(
+pub(crate) async fn run_turn(
     sess: Arc<Session>,
     turn_context: Arc<TurnContext>,
     input: Vec<UserInput>,
@@ -2313,7 +2342,7 @@ pub(crate) async fn run_task(
     if total_usage_tokens >= auto_compact_limit {
         run_auto_compact(&sess, &turn_context).await;
     }
-    let event = EventMsg::TaskStarted(TaskStartedEvent {
+    let event = EventMsg::TurnStarted(TurnStartedEvent {
         model_context_window: turn_context.client.get_model_context_window(),
     });
     sess.send_event(&turn_context, event).await;
@@ -2367,7 +2396,7 @@ pub(crate) async fn run_task(
         let turn_input: Vec<ResponseItem> = {
             sess.record_conversation_items(&turn_context, &pending_input)
                 .await;
-            sess.clone_history().await.get_history_for_prompt()
+            sess.clone_history().await.for_prompt()
         };
 
         let turn_input_messages = turn_input
@@ -2378,7 +2407,7 @@ pub(crate) async fn run_task(
             })
             .map(|user_message| user_message.message())
             .collect::<Vec<String>>();
-        match run_turn(
+        match run_model_turn(
             Arc::clone(&sess),
             Arc::clone(&turn_context),
             Arc::clone(&turn_diff_tracker),
@@ -2455,7 +2484,7 @@ async fn run_auto_compact(sess: &Arc<Session>, turn_context: &Arc<TurnContext>)
         cwd = %turn_context.cwd.display()
     )
 )]
-async fn run_turn(
+async fn run_model_turn(
     sess: Arc<Session>,
     turn_context: Arc<TurnContext>,
     turn_diff_tracker: SharedTurnDiffTracker,
@@ -2495,7 +2524,7 @@ async fn run_turn(
 
     let mut retries = 0;
     loop {
-        match try_run_turn(
+        let err = match try_run_turn(
             Arc::clone(&router),
             Arc::clone(&sess),
             Arc::clone(&turn_context),
@@ -2505,17 +2534,10 @@ async fn run_turn(
         )
         .await
         {
-            // todo(aibrahim): map special cases and ? on other errors
             Ok(output) => return Ok(output),
-            Err(CodexErr::TurnAborted) => {
-                return Err(CodexErr::TurnAborted);
-            }
-            Err(CodexErr::Interrupted) => return Err(CodexErr::Interrupted),
-            Err(CodexErr::EnvVar(var)) => return Err(CodexErr::EnvVar(var)),
-            Err(e @ CodexErr::Fatal(_)) => return Err(e),
-            Err(e @ CodexErr::ContextWindowExceeded) => {
+            Err(CodexErr::ContextWindowExceeded) => {
                 sess.set_total_tokens_full(&turn_context).await;
-                return Err(e);
+                return Err(CodexErr::ContextWindowExceeded);
             }
             Err(CodexErr::UsageLimitReached(e)) => {
                 let rate_limits = e.rate_limits.clone();
@@ -2524,39 +2546,38 @@ async fn run_turn(
                 }
                 return Err(CodexErr::UsageLimitReached(e));
             }
-            Err(CodexErr::UsageNotIncluded) => return Err(CodexErr::UsageNotIncluded),
-            Err(e @ CodexErr::QuotaExceeded) => return Err(e),
-            Err(e @ CodexErr::InvalidImageRequest()) => return Err(e),
-            Err(e @ CodexErr::InvalidRequest(_)) => return Err(e),
-            Err(e @ CodexErr::RefreshTokenFailed(_)) => return Err(e),
-            Err(e) => {
-                // Use the configured provider-specific stream retry budget.
-                let max_retries = turn_context.client.get_provider().stream_max_retries();
-                if retries < max_retries {
-                    retries += 1;
-                    let delay = match e {
-                        CodexErr::Stream(_, Some(delay)) => delay,
-                        _ => backoff(retries),
-                    };
-                    warn!(
-                        "stream disconnected - retrying turn ({retries}/{max_retries} in {delay:?})...",
-                    );
+            Err(err) => err,
+        };
 
-                    // Surface retry information to any UI/front‑end so the
-                    // user understands what is happening instead of staring
-                    // at a seemingly frozen screen.
-                    sess.notify_stream_error(
-                        &turn_context,
-                        format!("Reconnecting... {retries}/{max_retries}"),
-                        e,
-                    )
-                    .await;
+        if !err.is_retryable() {
+            return Err(err);
+        }
 
-                    tokio::time::sleep(delay).await;
-                } else {
-                    return Err(e);
+        // Use the configured provider-specific stream retry budget.
+        let max_retries = turn_context.client.get_provider().stream_max_retries();
+        if retries < max_retries {
+            retries += 1;
+            let delay = match &err {
+                CodexErr::Stream(_, requested_delay) => {
+                    requested_delay.unwrap_or_else(|| backoff(retries))
                 }
-            }
+                _ => backoff(retries),
+            };
+            warn!("stream disconnected - retrying turn ({retries}/{max_retries} in {delay:?})...",);
+
+            // Surface retry information to any UI/front‑end so the
+            // user understands what is happening instead of staring
+            // at a seemingly frozen screen.
+            sess.notify_stream_error(
+                &turn_context,
+                format!("Reconnecting... {retries}/{max_retries}"),
+                err,
+            )
+            .await;
+
+            tokio::time::sleep(delay).await;
+        } else {
+            return Err(err);
         }
     }
 }
@@ -2841,6 +2862,7 @@ pub(super) fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) -
 #[cfg(test)]
 pub(crate) use tests::make_session_and_context;
 
+use crate::git_info::get_git_repo_root;
 #[cfg(test)]
 pub(crate) use tests::make_session_and_context_with_rx;
 
@@ -2915,8 +2937,8 @@ mod tests {
             }))
             .await;
 
-        let actual = session.state.lock().await.clone_history().get_history();
-        assert_eq!(expected, actual);
+        let history = session.state.lock().await.clone_history();
+        assert_eq!(expected, history.raw_items());
     }
 
     #[tokio::test]
@@ -3005,8 +3027,8 @@ mod tests {
             .record_initial_history(InitialHistory::Forked(rollout_items))
             .await;
 
-        let actual = session.state.lock().await.clone_history().get_history();
-        assert_eq!(expected, actual);
+        let history = session.state.lock().await.clone_history();
+        assert_eq!(expected, history.raw_items());
     }
 
     #[tokio::test]
@@ -3062,8 +3084,8 @@ mod tests {
         expected.extend(initial_context);
         expected.extend(turn_1);
 
-        let actual = sess.clone_history().await.get_history();
-        assert_eq!(expected, actual);
+        let history = sess.clone_history().await;
+        assert_eq!(expected, history.raw_items());
     }
 
     #[tokio::test]
@@ -3088,8 +3110,8 @@ mod tests {
         let rollback_event = wait_for_thread_rolled_back(&rx).await;
         assert_eq!(rollback_event.num_turns, 99);
 
-        let actual = sess.clone_history().await.get_history();
-        assert_eq!(initial_context, actual);
+        let history = sess.clone_history().await;
+        assert_eq!(initial_context, history.raw_items());
     }
 
     #[tokio::test]
@@ -3109,8 +3131,8 @@ mod tests {
             Some(CodexErrorInfo::ThreadRollbackFailed)
         );
 
-        let actual = sess.clone_history().await.get_history();
-        assert_eq!(initial_context, actual);
+        let history = sess.clone_history().await;
+        assert_eq!(initial_context, history.raw_items());
     }
 
     #[tokio::test]
@@ -3130,8 +3152,8 @@ mod tests {
             Some(CodexErrorInfo::ThreadRollbackFailed)
         );
 
-        let actual = sess.clone_history().await.get_history();
-        assert_eq!(initial_context, actual);
+        let history = sess.clone_history().await;
+        assert_eq!(initial_context, history.raw_items());
     }
 
     #[tokio::test]
@@ -3632,8 +3654,8 @@ mod tests {
             .record_model_warning("too many unified exec processes", &turn_context)
             .await;
 
-        let mut history = session.clone_history().await;
-        let history_items = history.get_history();
+        let history = session.clone_history().await;
+        let history_items = history.raw_items();
         let last = history_items.last().expect("warning recorded");
 
         match last {
@@ -3779,8 +3801,9 @@ mod tests {
             }
         }
 
-        let history = sess.clone_history().await.get_history();
-        let _ = history;
+        // TODO(jif) investigate what is this?
+        let history = sess.clone_history().await;
+        let _ = history.raw_items();
     }
 
     #[tokio::test]
@@ -3869,7 +3892,7 @@ mod tests {
         rollout_items.push(RolloutItem::ResponseItem(assistant1.clone()));
 
         let summary1 = "summary one";
-        let snapshot1 = live_history.get_history();
+        let snapshot1 = live_history.clone().for_prompt();
         let user_messages1 = collect_user_messages(&snapshot1);
         let rebuilt1 = compact::build_compacted_history(
             session.build_initial_context(turn_context),
@@ -3903,7 +3926,7 @@ mod tests {
         rollout_items.push(RolloutItem::ResponseItem(assistant2.clone()));
 
         let summary2 = "summary two";
-        let snapshot2 = live_history.get_history();
+        let snapshot2 = live_history.clone().for_prompt();
         let user_messages2 = collect_user_messages(&snapshot2);
         let rebuilt2 = compact::build_compacted_history(
             session.build_initial_context(turn_context),
@@ -3936,7 +3959,7 @@ mod tests {
         live_history.record_items(std::iter::once(&assistant3), turn_context.truncation_policy);
         rollout_items.push(RolloutItem::ResponseItem(assistant3.clone()));
 
-        (rollout_items, live_history.get_history())
+        (rollout_items, live_history.for_prompt())
     }
 
     #[tokio::test]

codex-rs/core/src/codex_delegate.rs

@@ -135,7 +135,7 @@ pub(crate) async fn run_codex_thread_one_shot(
         while let Ok(event) = io_for_bridge.next_event().await {
             let should_shutdown = matches!(
                 event.msg,
-                EventMsg::TaskComplete(_) | EventMsg::TurnAborted(_)
+                EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)
             );
             let _ = tx_bridge.send(event).await;
             if should_shutdown {
@@ -253,7 +253,7 @@ async fn shutdown_delegate(codex: &Codex) {
         while let Ok(event) = codex.next_event().await {
             if matches!(
                 event.msg,
-                EventMsg::TurnAborted(_) | EventMsg::TaskComplete(_)
+                EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_)
             ) {
                 break;
             }

codex-rs/core/src/compact.rs

@@ -12,8 +12,8 @@ use crate::features::Feature;
 use crate::protocol::CompactedItem;
 use crate::protocol::ContextCompactedEvent;
 use crate::protocol::EventMsg;
-use crate::protocol::TaskStartedEvent;
 use crate::protocol::TurnContextItem;
+use crate::protocol::TurnStartedEvent;
 use crate::protocol::WarningEvent;
 use crate::truncate::TruncationPolicy;
 use crate::truncate::approx_token_count;
@@ -54,7 +54,7 @@ pub(crate) async fn run_compact_task(
     turn_context: Arc<TurnContext>,
     input: Vec<UserInput>,
 ) {
-    let start_event = EventMsg::TaskStarted(TaskStartedEvent {
+    let start_event = EventMsg::TurnStarted(TurnStartedEvent {
         model_context_window: turn_context.client.get_model_context_window(),
     });
     sess.send_event(&turn_context, start_event).await;
@@ -95,9 +95,11 @@ async fn run_compact_task_inner(
     sess.persist_rollout_items(&[rollout_item]).await;
 
     loop {
-        let turn_input = history.get_history_for_prompt();
+        // Clone is required because of the loop
+        let turn_input = history.clone().for_prompt();
+        let turn_input_len = turn_input.len();
         let prompt = Prompt {
-            input: turn_input.clone(),
+            input: turn_input,
             ..Default::default()
         };
         let attempt_result = drain_to_completed(&sess, turn_context.as_ref(), &prompt).await;
@@ -119,7 +121,7 @@ async fn run_compact_task_inner(
                 return;
             }
             Err(e @ CodexErr::ContextWindowExceeded) => {
-                if turn_input.len() > 1 {
+                if turn_input_len > 1 {
                     // Trim from the beginning to preserve cache (prefix-based) and keep recent messages intact.
                     error!(
                         "Context window exceeded while compacting; removing oldest history item. Error: {e}"
@@ -155,15 +157,15 @@ async fn run_compact_task_inner(
         }
     }
 
-    let history_snapshot = sess.clone_history().await.get_history();
-    let summary_suffix =
-        get_last_assistant_message_from_turn(&history_snapshot).unwrap_or_default();
+    let history_snapshot = sess.clone_history().await;
+    let history_items = history_snapshot.raw_items();
+    let summary_suffix = get_last_assistant_message_from_turn(history_items).unwrap_or_default();
     let summary_text = format!("{SUMMARY_PREFIX}\n{summary_suffix}");
-    let user_messages = collect_user_messages(&history_snapshot);
+    let user_messages = collect_user_messages(history_items);
 
     let initial_context = sess.build_initial_context(turn_context.as_ref());
     let mut new_history = build_compacted_history(initial_context, &user_messages, &summary_text);
-    let ghost_snapshots: Vec<ResponseItem> = history_snapshot
+    let ghost_snapshots: Vec<ResponseItem> = history_items
         .iter()
         .filter(|item| matches!(item, ResponseItem::GhostSnapshot { .. }))
         .cloned()

codex-rs/core/src/compact_remote.rs

@@ -8,7 +8,7 @@ use crate::protocol::CompactedItem;
 use crate::protocol::ContextCompactedEvent;
 use crate::protocol::EventMsg;
 use crate::protocol::RolloutItem;
-use crate::protocol::TaskStartedEvent;
+use crate::protocol::TurnStartedEvent;
 use codex_protocol::models::ResponseItem;
 
 pub(crate) async fn run_inline_remote_auto_compact_task(
@@ -19,7 +19,7 @@ pub(crate) async fn run_inline_remote_auto_compact_task(
 }
 
 pub(crate) async fn run_remote_compact_task(sess: Arc<Session>, turn_context: Arc<TurnContext>) {
-    let start_event = EventMsg::TaskStarted(TaskStartedEvent {
+    let start_event = EventMsg::TurnStarted(TurnStartedEvent {
         model_context_window: turn_context.client.get_model_context_window(),
     });
     sess.send_event(&turn_context, start_event).await;
@@ -40,9 +40,18 @@ async fn run_remote_compact_task_inner_impl(
     sess: &Arc<Session>,
     turn_context: &Arc<TurnContext>,
 ) -> CodexResult<()> {
-    let mut history = sess.clone_history().await;
+    let history = sess.clone_history().await;
+
+    // Required to keep `/undo` available after compaction
+    let ghost_snapshots: Vec<ResponseItem> = history
+        .raw_items()
+        .iter()
+        .filter(|item| matches!(item, ResponseItem::GhostSnapshot { .. }))
+        .cloned()
+        .collect();
+
     let prompt = Prompt {
-        input: history.get_history_for_prompt(),
+        input: history.for_prompt(),
         tools: vec![],
         parallel_tool_calls: false,
         base_instructions_override: turn_context.base_instructions.clone(),
@@ -53,13 +62,6 @@ async fn run_remote_compact_task_inner_impl(
         .client
         .compact_conversation_history(&prompt)
         .await?;
-    // Required to keep `/undo` available after compaction
-    let ghost_snapshots: Vec<ResponseItem> = history
-        .get_history()
-        .iter()
-        .filter(|item| matches!(item, ResponseItem::GhostSnapshot { .. }))
-        .cloned()
-        .collect();
 
     if !ghost_snapshots.is_empty() {
         new_history.extend(ghost_snapshots);