changes

Adds a TUI for managing, applying, and creating cloud tasks

.codespellrc

@@ -1,6 +1,6 @@
 [codespell]
 # Ref: https://github.com/codespell-project/codespell#using-a-config-file
-skip = .git*,vendor,*-lock.yaml,*.lock,.codespellrc,*test.ts,*.jsonl
+skip = .git*,vendor,*-lock.yaml,*.lock,.codespellrc,*test.ts,*.jsonl,frame*.txt
 check-hidden = true
 ignore-regex = ^\s*"image/\S+": ".*|\b(afterAll)\b
 ignore-words-list = ratatui,ser

.github/dotslash-config.json

@@ -27,6 +27,34 @@
           "path": "codex.exe"
         }
       }
+    },
+    "codex-responses-api-proxy": {
+      "platforms": {
+        "macos-aarch64": {
+          "regex": "^codex-responses-api-proxy-aarch64-apple-darwin\\.zst$",
+          "path": "codex-responses-api-proxy"
+        },
+        "macos-x86_64": {
+          "regex": "^codex-responses-api-proxy-x86_64-apple-darwin\\.zst$",
+          "path": "codex-responses-api-proxy"
+        },
+        "linux-x86_64": {
+          "regex": "^codex-responses-api-proxy-x86_64-unknown-linux-musl\\.zst$",
+          "path": "codex-responses-api-proxy"
+        },
+        "linux-aarch64": {
+          "regex": "^codex-responses-api-proxy-aarch64-unknown-linux-musl\\.zst$",
+          "path": "codex-responses-api-proxy"
+        },
+        "windows-x86_64": {
+          "regex": "^codex-responses-api-proxy-x86_64-pc-windows-msvc\\.exe\\.zst$",
+          "path": "codex-responses-api-proxy.exe"
+        },
+        "windows-aarch64": {
+          "regex": "^codex-responses-api-proxy-aarch64-pc-windows-msvc\\.exe\\.zst$",
+          "path": "codex-responses-api-proxy.exe"
+        }
+      }
     }
   }
 }

.github/workflows/ci.yml

@@ -1,7 +1,7 @@
 name: ci
 
 on:
-  pull_request: { branches: [main] }
+  pull_request: {}
   push: { branches: [main] }
 
 jobs:
@@ -31,6 +31,7 @@ jobs:
       - uses: facebook/install-dotslash@v2
 
       - name: Stage npm package
+        id: stage_npm_package
         env:
           GH_TOKEN: ${{ github.token }}
         run: |
@@ -40,13 +41,13 @@ jobs:
           python3 ./codex-cli/scripts/build_npm_package.py \
             --release-version "$CODEX_VERSION" \
             --pack-output "$PACK_OUTPUT"
-          echo "PACK_OUTPUT=$PACK_OUTPUT" >> "$GITHUB_ENV"
+          echo "pack_output=$PACK_OUTPUT" >> "$GITHUB_OUTPUT"
 
       - name: Upload staged npm package artifact
         uses: actions/upload-artifact@v4
         with:
           name: codex-npm-staging
-          path: ${{ env.PACK_OUTPUT }}
+          path: ${{ steps.stage_npm_package.outputs.pack_output }}
 
       - name: Ensure root README.md contains only ASCII and certain Unicode code points
         run: ./scripts/asciicheck.py README.md

.github/workflows/codespell.yml

@@ -22,7 +22,6 @@ jobs:
       - name: Annotate locations with typos
         uses: codespell-project/codespell-problem-matcher@b80729f885d32f78a716c2f107b4db1025001c42 # v1
       - name: Codespell
-        uses: codespell-project/actions-codespell@406322ec52dd7b488e48c1c4b82e2a8b3a1bf630 # v2
+        uses: codespell-project/actions-codespell@406322ec52dd7b488e48c1c4b82e2a8b3a1bf630 # v2.1
         with:
           ignore_words_file: .codespellignore
-          skip: frame*.txt

.github/workflows/rust-ci.yml

@@ -57,7 +57,7 @@ jobs:
         working-directory: codex-rs
     steps:
       - uses: actions/checkout@v5
-      - uses: dtolnay/rust-toolchain@1.89
+      - uses: dtolnay/rust-toolchain@1.90
         with:
           components: rustfmt
       - name: cargo fmt
@@ -75,7 +75,7 @@ jobs:
         working-directory: codex-rs
     steps:
       - uses: actions/checkout@v5
-      - uses: dtolnay/rust-toolchain@1.89
+      - uses: dtolnay/rust-toolchain@1.90
       - uses: taiki-e/install-action@0c5db7f7f897c03b771660e91d065338615679f4 # v2
         with:
           tool: cargo-shear
@@ -143,7 +143,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v5
-      - uses: dtolnay/rust-toolchain@1.89
+      - uses: dtolnay/rust-toolchain@1.90
         with:
           targets: ${{ matrix.target }}
           components: clippy

.github/workflows/rust-release.yml

@@ -77,7 +77,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v5
-      - uses: dtolnay/rust-toolchain@1.89
+      - uses: dtolnay/rust-toolchain@1.90
         with:
           targets: ${{ matrix.target }}
 
@@ -97,7 +97,7 @@ jobs:
           sudo apt install -y musl-tools pkg-config
 
       - name: Cargo build
-        run: cargo build --target ${{ matrix.target }} --release --bin codex
+        run: cargo build --target ${{ matrix.target }} --release --bin codex --bin codex-responses-api-proxy
 
       - name: Stage artifacts
         shell: bash
@@ -107,8 +107,10 @@ jobs:
 
           if [[ "${{ matrix.runner }}" == windows* ]]; then
             cp target/${{ matrix.target }}/release/codex.exe "$dest/codex-${{ matrix.target }}.exe"
+            cp target/${{ matrix.target }}/release/codex-responses-api-proxy.exe "$dest/codex-responses-api-proxy-${{ matrix.target }}.exe"
           else
             cp target/${{ matrix.target }}/release/codex "$dest/codex-${{ matrix.target }}"
+            cp target/${{ matrix.target }}/release/codex-responses-api-proxy "$dest/codex-responses-api-proxy-${{ matrix.target }}"
           fi
 
       - if: ${{ matrix.runner == 'windows-11-arm' }}
@@ -216,17 +218,30 @@ jobs:
 
       # build_npm_package.py requires DotSlash when staging releases.
       - uses: facebook/install-dotslash@v2
-      - name: Stage npm package
+      - name: Stage codex CLI npm package
         env:
           GH_TOKEN: ${{ github.token }}
         run: |
           set -euo pipefail
           TMP_DIR="${RUNNER_TEMP}/npm-stage"
           ./codex-cli/scripts/build_npm_package.py \
+            --package codex \
             --release-version "${{ steps.release_name.outputs.name }}" \
             --staging-dir "${TMP_DIR}" \
             --pack-output "${GITHUB_WORKSPACE}/dist/npm/codex-npm-${{ steps.release_name.outputs.name }}.tgz"
 
+      - name: Stage responses API proxy npm package
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          set -euo pipefail
+          TMP_DIR="${RUNNER_TEMP}/npm-stage-responses"
+          ./codex-cli/scripts/build_npm_package.py \
+            --package codex-responses-api-proxy \
+            --release-version "${{ steps.release_name.outputs.name }}" \
+            --staging-dir "${TMP_DIR}" \
+            --pack-output "${GITHUB_WORKSPACE}/dist/npm/codex-responses-api-proxy-npm-${{ steps.release_name.outputs.name }}.tgz"
+
       - name: Create GitHub Release
         uses: softprops/action-gh-release@v2
         with:
@@ -269,7 +284,7 @@ jobs:
       - name: Update npm
         run: npm install -g npm@latest
 
-      - name: Download npm tarball from release
+      - name: Download npm tarballs from release
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         run: |
@@ -281,6 +296,10 @@ jobs:
             --repo "${GITHUB_REPOSITORY}" \
             --pattern "codex-npm-${version}.tgz" \
             --dir dist/npm
+          gh release download "$tag" \
+            --repo "${GITHUB_REPOSITORY}" \
+            --pattern "codex-responses-api-proxy-npm-${version}.tgz" \
+            --dir dist/npm
 
       # No NODE_AUTH_TOKEN needed because we use OIDC.
       - name: Publish to npm
@@ -294,7 +313,14 @@ jobs:
             tag_args+=(--tag "${NPM_TAG}")
           fi
 
-          npm publish "${GITHUB_WORKSPACE}/dist/npm/codex-npm-${VERSION}.tgz" "${tag_args[@]}"
+          tarballs=(
+            "codex-npm-${VERSION}.tgz"
+            "codex-responses-api-proxy-npm-${VERSION}.tgz"
+          )
+
+          for tarball in "${tarballs[@]}"; do
+            npm publish "${GITHUB_WORKSPACE}/dist/npm/${tarball}" "${tag_args[@]}"
+          done
 
   update-branch:
     name: Update latest-alpha-cli branch

.github/workflows/sdk.yml

@@ -0,0 +1,43 @@
+name: sdk
+
+on:
+  push:
+    branches: [main]
+  pull_request: {}
+
+jobs:
+  sdks:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v5
+
+      - name: Setup pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          run_install: false
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v5
+        with:
+          node-version: 22
+          cache: pnpm
+
+      - uses: dtolnay/rust-toolchain@1.90
+
+      - name: build codex
+        run: cargo build --bin codex
+        working-directory: codex-rs
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Build SDK packages
+        run: pnpm -r --filter ./sdk/typescript run build
+
+      - name: Lint SDK packages
+        run: pnpm -r --filter ./sdk/typescript run lint
+
+      - name: Test SDK packages
+        run: pnpm -r --filter ./sdk/typescript run test

README.md

@@ -1,4 +1,3 @@
-<h1 align="center">OpenAI Codex CLI</h1>
 
 <p align="center"><code>npm i -g @openai/codex</code><br />or <code>brew install codex</code></p>
 
@@ -102,4 +101,3 @@ Codex CLI supports a rich set of configuration options, with preferences stored
 ## License
 
 This repository is licensed under the [Apache-2.0 License](LICENSE).
-

codex-cli/README.md

@@ -208,7 +208,7 @@ The hardening mechanism Codex uses depends on your OS:
 | Requirement                 | Details                                                         |
 | --------------------------- | --------------------------------------------------------------- |
 | Operating systems           | macOS 12+, Ubuntu 20.04+/Debian 10+, or Windows 11 **via WSL2** |
-| Node.js                     | **22 or newer** (LTS recommended)                               |
+| Node.js                     | **16 or newer** (Node 20 LTS recommended)                       |
 | Git (optional, recommended) | 2.23+ for built-in PR helpers                                   |
 | RAM                         | 4-GB minimum (8-GB recommended)                                 |
 
@@ -513,7 +513,7 @@ Codex runs model-generated commands in a sandbox. If a proposed command or file
 <details>
 <summary>Does it work on Windows?</summary>
 
-Not directly. It requires [Windows Subsystem for Linux (WSL2)](https://learn.microsoft.com/en-us/windows/wsl/install) - Codex has been tested on macOS and Linux with Node 22.
+Not directly. It requires [Windows Subsystem for Linux (WSL2)](https://learn.microsoft.com/en-us/windows/wsl/install) - Codex is regularly tested on macOS and Linux with Node 20+, and also supports Node 16.
 
 </details>
 

codex-cli/bin/codex.js

@@ -1,6 +1,7 @@
 #!/usr/bin/env node
 // Unified entry point for the Codex CLI.
 
+import { spawn } from "node:child_process";
 import { existsSync } from "fs";
 import path from "path";
 import { fileURLToPath } from "url";
@@ -68,7 +69,6 @@ const binaryPath = path.join(archRoot, "codex", codexBinaryName);
 // executing. This allows us to forward those signals to the child process
 // and guarantees that when either the child terminates or the parent
 // receives a fatal signal, both processes exit in a predictable manner.
-const { spawn } = await import("child_process");
 
 function getUpdatedPath(newDirs) {
   const pathSep = process.platform === "win32" ? ";" : ":";

codex-cli/package-lock.json

@@ -11,7 +11,7 @@
         "codex": "bin/codex.js"
       },
       "engines": {
-        "node": ">=20"
+        "node": ">=16"
       }
     }
   }

codex-cli/package.json

@@ -7,7 +7,7 @@
   },
   "type": "module",
   "engines": {
-    "node": ">=20"
+    "node": ">=16"
   },
   "files": [
     "bin",

codex-cli/scripts/build_npm_package.py

@@ -13,11 +13,23 @@ from pathlib import Path
 SCRIPT_DIR = Path(__file__).resolve().parent
 CODEX_CLI_ROOT = SCRIPT_DIR.parent
 REPO_ROOT = CODEX_CLI_ROOT.parent
+RESPONSES_API_PROXY_NPM_ROOT = REPO_ROOT / "codex-rs" / "responses-api-proxy" / "npm"
 GITHUB_REPO = "openai/codex"
 
+# The docs are not clear on what the expected value/format of
+# workflow/workflowName is:
+# https://cli.github.com/manual/gh_run_list
+WORKFLOW_NAME = ".github/workflows/rust-release.yml"
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Build or stage the Codex CLI npm package.")
+    parser.add_argument(
+        "--package",
+        choices=("codex", "codex-responses-api-proxy"),
+        default="codex",
+        help="Which npm package to stage (default: codex).",
+    )
     parser.add_argument(
         "--version",
         help="Version number to write to package.json inside the staged package.",
@@ -58,6 +70,7 @@ def parse_args() -> argparse.Namespace:
 def main() -> int:
     args = parse_args()
 
+    package = args.package
     version = args.version
     release_version = args.release_version
     if release_version:
@@ -71,7 +84,7 @@ def main() -> int:
     staging_dir, created_temp = prepare_staging_dir(args.staging_dir)
 
     try:
-        stage_sources(staging_dir, version)
+        stage_sources(staging_dir, version, package)
 
         workflow_url = args.workflow_url
         resolved_head_sha: str | None = None
@@ -95,16 +108,23 @@ def main() -> int:
         if not workflow_url:
             raise RuntimeError("Unable to determine workflow URL for native binaries.")
 
-        install_native_binaries(staging_dir, workflow_url)
+        install_native_binaries(staging_dir, workflow_url, package)
 
         if release_version:
             staging_dir_str = str(staging_dir)
-            print(
-                f"Staged version {version} for release in {staging_dir_str}\n\n"
-                "Verify the CLI:\n"
-                f"    node {staging_dir_str}/bin/codex.js --version\n"
-                f"    node {staging_dir_str}/bin/codex.js --help\n\n"
-            )
+            if package == "codex":
+                print(
+                    f"Staged version {version} for release in {staging_dir_str}\n\n"
+                    "Verify the CLI:\n"
+                    f"    node {staging_dir_str}/bin/codex.js --version\n"
+                    f"    node {staging_dir_str}/bin/codex.js --help\n\n"
+                )
+            else:
+                print(
+                    f"Staged version {version} for release in {staging_dir_str}\n\n"
+                    "Verify the responses API proxy:\n"
+                    f"    node {staging_dir_str}/bin/codex-responses-api-proxy.js --help\n\n"
+                )
         else:
             print(f"Staged package in {staging_dir}")
 
@@ -131,20 +151,34 @@ def prepare_staging_dir(staging_dir: Path | None) -> tuple[Path, bool]:
     return temp_dir, True
 
 
-def stage_sources(staging_dir: Path, version: str) -> None:
+def stage_sources(staging_dir: Path, version: str, package: str) -> None:
     bin_dir = staging_dir / "bin"
     bin_dir.mkdir(parents=True, exist_ok=True)
 
-    shutil.copy2(CODEX_CLI_ROOT / "bin" / "codex.js", bin_dir / "codex.js")
-    rg_manifest = CODEX_CLI_ROOT / "bin" / "rg"
-    if rg_manifest.exists():
-        shutil.copy2(rg_manifest, bin_dir / "rg")
+    if package == "codex":
+        shutil.copy2(CODEX_CLI_ROOT / "bin" / "codex.js", bin_dir / "codex.js")
+        rg_manifest = CODEX_CLI_ROOT / "bin" / "rg"
+        if rg_manifest.exists():
+            shutil.copy2(rg_manifest, bin_dir / "rg")
 
-    readme_src = REPO_ROOT / "README.md"
-    if readme_src.exists():
-        shutil.copy2(readme_src, staging_dir / "README.md")
+        readme_src = REPO_ROOT / "README.md"
+        if readme_src.exists():
+            shutil.copy2(readme_src, staging_dir / "README.md")
 
-    with open(CODEX_CLI_ROOT / "package.json", "r", encoding="utf-8") as fh:
+        package_json_path = CODEX_CLI_ROOT / "package.json"
+    elif package == "codex-responses-api-proxy":
+        launcher_src = RESPONSES_API_PROXY_NPM_ROOT / "bin" / "codex-responses-api-proxy.js"
+        shutil.copy2(launcher_src, bin_dir / "codex-responses-api-proxy.js")
+
+        readme_src = RESPONSES_API_PROXY_NPM_ROOT / "README.md"
+        if readme_src.exists():
+            shutil.copy2(readme_src, staging_dir / "README.md")
+
+        package_json_path = RESPONSES_API_PROXY_NPM_ROOT / "package.json"
+    else:
+        raise RuntimeError(f"Unknown package '{package}'.")
+
+    with open(package_json_path, "r", encoding="utf-8") as fh:
         package_json = json.load(fh)
     package_json["version"] = version
 
@@ -153,20 +187,27 @@ def stage_sources(staging_dir: Path, version: str) -> None:
         out.write("\n")
 
 
-def install_native_binaries(staging_dir: Path, workflow_url: str | None) -> None:
-    cmd = ["./scripts/install_native_deps.py"]
-    if workflow_url:
-        cmd.extend(["--workflow-url", workflow_url])
+def install_native_binaries(staging_dir: Path, workflow_url: str, package: str) -> None:
+    package_components = {
+        "codex": ["codex", "rg"],
+        "codex-responses-api-proxy": ["codex-responses-api-proxy"],
+    }
+
+    components = package_components.get(package)
+    if components is None:
+        raise RuntimeError(f"Unknown package '{package}'.")
+
+    cmd = ["./scripts/install_native_deps.py", "--workflow-url", workflow_url]
+    for component in components:
+        cmd.extend(["--component", component])
     cmd.append(str(staging_dir))
     subprocess.check_call(cmd, cwd=CODEX_CLI_ROOT)
 
 
 def resolve_latest_alpha_workflow_url() -> str:
     version = determine_latest_alpha_version()
-    workflow_url = fetch_workflow_url_for_version(version)
-    if not workflow_url:
-        raise RuntimeError(f"Unable to locate workflow for version {version}.")
-    return workflow_url
+    workflow = resolve_release_workflow(version)
+    return workflow["url"]
 
 
 def determine_latest_alpha_version() -> str:
@@ -205,36 +246,6 @@ def list_releases() -> list[dict]:
     return releases
 
 
-def fetch_workflow_url_for_version(version: str) -> str | None:
-    ref = f"rust-v{version}"
-    stdout = subprocess.check_output(
-        [
-            "gh",
-            "run",
-            "list",
-            "--branch",
-            ref,
-            "--limit",
-            "20",
-            "--json",
-            "workflowName,url",
-        ],
-        text=True,
-    )
-
-    try:
-        runs = json.loads(stdout or "[]")
-    except json.JSONDecodeError as exc:
-        raise RuntimeError("Unable to parse workflow run listing.") from exc
-
-    for run in runs:
-        if run.get("workflowName") == "rust-release":
-            url = run.get("url")
-            if url:
-                return url
-    return None
-
-
 def resolve_release_workflow(version: str) -> dict:
     stdout = subprocess.check_output(
         [
@@ -245,12 +256,14 @@ def resolve_release_workflow(version: str) -> dict:
             f"rust-v{version}",
             "--json",
             "workflowName,url,headSha",
+            "--workflow",
+            WORKFLOW_NAME,
             "--jq",
-            'first(.[] | select(.workflowName == "rust-release"))',
+            "first(.[])",
         ],
         text=True,
     )
-    workflow = json.loads(stdout)
+    workflow = json.loads(stdout or "[]")
     if not workflow:
         raise RuntimeError(f"Unable to find rust-release workflow for version {version}.")
     return workflow

codex-cli/scripts/install_native_deps.py

@@ -9,6 +9,7 @@ import subprocess
 import tarfile
 import tempfile
 import zipfile
+from dataclasses import dataclass
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 from typing import Iterable, Sequence
@@ -20,7 +21,7 @@ CODEX_CLI_ROOT = SCRIPT_DIR.parent
 DEFAULT_WORKFLOW_URL = "https://github.com/openai/codex/actions/runs/17952349351"  # rust-v0.40.0
 VENDOR_DIR_NAME = "vendor"
 RG_MANIFEST = CODEX_CLI_ROOT / "bin" / "rg"
-CODEX_TARGETS = (
+BINARY_TARGETS = (
     "x86_64-unknown-linux-musl",
     "aarch64-unknown-linux-musl",
     "x86_64-apple-darwin",
@@ -29,6 +30,27 @@ CODEX_TARGETS = (
     "aarch64-pc-windows-msvc",
 )
 
+
+@dataclass(frozen=True)
+class BinaryComponent:
+    artifact_prefix: str  # matches the artifact filename prefix (e.g. codex-<target>.zst)
+    dest_dir: str  # directory under vendor/<target>/ where the binary is installed
+    binary_basename: str  # executable name inside dest_dir (before optional .exe)
+
+
+BINARY_COMPONENTS = {
+    "codex": BinaryComponent(
+        artifact_prefix="codex",
+        dest_dir="codex",
+        binary_basename="codex",
+    ),
+    "codex-responses-api-proxy": BinaryComponent(
+        artifact_prefix="codex-responses-api-proxy",
+        dest_dir="codex-responses-api-proxy",
+        binary_basename="codex-responses-api-proxy",
+    ),
+}
+
 RG_TARGET_PLATFORM_PAIRS: list[tuple[str, str]] = [
     ("x86_64-unknown-linux-musl", "linux-x86_64"),
     ("aarch64-unknown-linux-musl", "linux-aarch64"),
@@ -50,6 +72,16 @@ def parse_args() -> argparse.Namespace:
             "known good run when omitted."
         ),
     )
+    parser.add_argument(
+        "--component",
+        dest="components",
+        action="append",
+        choices=tuple(list(BINARY_COMPONENTS) + ["rg"]),
+        help=(
+            "Limit installation to the specified components."
+            " May be repeated. Defaults to 'codex' and 'rg'."
+        ),
+    )
     parser.add_argument(
         "root",
         nargs="?",
@@ -69,18 +101,28 @@ def main() -> int:
     vendor_dir = codex_cli_root / VENDOR_DIR_NAME
     vendor_dir.mkdir(parents=True, exist_ok=True)
 
+    components = args.components or ["codex", "rg"]
+
     workflow_url = (args.workflow_url or DEFAULT_WORKFLOW_URL).strip()
     if not workflow_url:
         workflow_url = DEFAULT_WORKFLOW_URL
 
     workflow_id = workflow_url.rstrip("/").split("/")[-1]
+    print(f"Downloading native artifacts from workflow {workflow_id}...")
 
     with tempfile.TemporaryDirectory(prefix="codex-native-artifacts-") as artifacts_dir_str:
         artifacts_dir = Path(artifacts_dir_str)
         _download_artifacts(workflow_id, artifacts_dir)
-        install_codex_binaries(artifacts_dir, vendor_dir, CODEX_TARGETS)
+        install_binary_components(
+            artifacts_dir,
+            vendor_dir,
+            BINARY_TARGETS,
+            [name for name in components if name in BINARY_COMPONENTS],
+        )
 
-    fetch_rg(vendor_dir, DEFAULT_RG_TARGETS, manifest_path=RG_MANIFEST)
+    if "rg" in components:
+        print("Fetching ripgrep binaries...")
+        fetch_rg(vendor_dir, DEFAULT_RG_TARGETS, manifest_path=RG_MANIFEST)
 
     print(f"Installed native dependencies into {vendor_dir}")
     return 0
@@ -124,6 +166,8 @@ def fetch_rg(
     results: dict[str, Path] = {}
     max_workers = min(len(task_configs), max(1, (os.cpu_count() or 1)))
 
+    print("Installing ripgrep binaries for targets: " + ", ".join(targets))
+
     with ThreadPoolExecutor(max_workers=max_workers) as executor:
         future_map = {
             executor.submit(
@@ -140,6 +184,7 @@ def fetch_rg(
         for future in as_completed(future_map):
             target = future_map[future]
             results[target] = future.result()
+            print(f"  installed ripgrep for {target}")
 
     return [results[target] for target in targets]
 
@@ -158,40 +203,60 @@ def _download_artifacts(workflow_id: str, dest_dir: Path) -> None:
     subprocess.check_call(cmd)
 
 
-def install_codex_binaries(
-    artifacts_dir: Path, vendor_dir: Path, targets: Iterable[str]
-) -> list[Path]:
+def install_binary_components(
+    artifacts_dir: Path,
+    vendor_dir: Path,
+    targets: Iterable[str],
+    component_names: Sequence[str],
+) -> None:
+    selected_components = [BINARY_COMPONENTS[name] for name in component_names if name in BINARY_COMPONENTS]
+    if not selected_components:
+        return
+
     targets = list(targets)
     if not targets:
-        return []
+        return
 
-    results: dict[str, Path] = {}
-    max_workers = min(len(targets), max(1, (os.cpu_count() or 1)))
-
-    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        future_map = {
-            executor.submit(_install_single_codex_binary, artifacts_dir, vendor_dir, target): target
-            for target in targets
-        }
-
-        for future in as_completed(future_map):
-            target = future_map[future]
-            results[target] = future.result()
-
-    return [results[target] for target in targets]
+    for component in selected_components:
+        print(
+            f"Installing {component.binary_basename} binaries for targets: "
+            + ", ".join(targets)
+        )
+        max_workers = min(len(targets), max(1, (os.cpu_count() or 1)))
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            futures = {
+                executor.submit(
+                    _install_single_binary,
+                    artifacts_dir,
+                    vendor_dir,
+                    target,
+                    component,
+                ): target
+                for target in targets
+            }
+            for future in as_completed(futures):
+                installed_path = future.result()
+                print(f"  installed {installed_path}")
 
 
-def _install_single_codex_binary(artifacts_dir: Path, vendor_dir: Path, target: str) -> Path:
+def _install_single_binary(
+    artifacts_dir: Path,
+    vendor_dir: Path,
+    target: str,
+    component: BinaryComponent,
+) -> Path:
     artifact_subdir = artifacts_dir / target
-    archive_name = _archive_name_for_target(target)
+    archive_name = _archive_name_for_target(component.artifact_prefix, target)
     archive_path = artifact_subdir / archive_name
     if not archive_path.exists():
         raise FileNotFoundError(f"Expected artifact not found: {archive_path}")
 
-    dest_dir = vendor_dir / target / "codex"
+    dest_dir = vendor_dir / target / component.dest_dir
     dest_dir.mkdir(parents=True, exist_ok=True)
 
-    binary_name = "codex.exe" if "windows" in target else "codex"
+    binary_name = (
+        f"{component.binary_basename}.exe" if "windows" in target else component.binary_basename
+    )
     dest = dest_dir / binary_name
     dest.unlink(missing_ok=True)
     extract_archive(archive_path, "zst", None, dest)
@@ -200,10 +265,10 @@ def _install_single_codex_binary(artifacts_dir: Path, vendor_dir: Path, target:
     return dest
 
 
-def _archive_name_for_target(target: str) -> str:
+def _archive_name_for_target(artifact_prefix: str, target: str) -> str:
     if "windows" in target:
-        return f"codex-{target}.exe.zst"
-    return f"codex-{target}.zst"
+        return f"{artifact_prefix}-{target}.exe.zst"
+    return f"{artifact_prefix}-{target}.zst"
 
 
 def _fetch_single_rg(

codex-rs/Cargo.toml

@@ -1,8 +1,13 @@
 [workspace]
 members = [
+    "backend-client",
     "ansi-escape",
+    "app-server",
     "apply-patch",
     "arg0",
+    "codex-backend-openapi-models",
+    "cloud-tasks",
+    "cloud-tasks-client",
     "cli",
     "common",
     "core",
@@ -16,9 +21,15 @@ members = [
     "mcp-server",
     "mcp-types",
     "ollama",
+    "process-hardening",
     "protocol",
     "protocol-ts",
+    "rmcp-client",
+    "responses-api-proxy",
+    "otel",
     "tui",
+    "git-apply",
+    "utils/json-to-toml",
     "utils/readiness",
 ]
 resolver = "2"
@@ -33,7 +44,9 @@ edition = "2024"
 
 [workspace.dependencies]
 # Internal
+app_test_support = { path = "app-server/tests/common" }
 codex-ansi-escape = { path = "ansi-escape" }
+codex-app-server = { path = "app-server" }
 codex-apply-patch = { path = "apply-patch" }
 codex-arg0 = { path = "arg0" }
 codex-chatgpt = { path = "chatgpt" }
@@ -47,9 +60,14 @@ codex-login = { path = "login" }
 codex-mcp-client = { path = "mcp-client" }
 codex-mcp-server = { path = "mcp-server" }
 codex-ollama = { path = "ollama" }
+codex-otel = { path = "otel" }
+codex-process-hardening = { path = "process-hardening" }
 codex-protocol = { path = "protocol" }
 codex-protocol-ts = { path = "protocol-ts" }
+codex-responses-api-proxy = { path = "responses-api-proxy" }
+codex-rmcp-client = { path = "rmcp-client" }
 codex-tui = { path = "tui" }
+codex-utils-json-to-toml = { path = "utils/json-to-toml" }
 codex-utils-readiness = { path = "utils/readiness" }
 core_test_support = { path = "core/tests/common" }
 mcp-types = { path = "mcp-types" }
@@ -67,23 +85,26 @@ async-stream = "0.3.6"
 async-trait = "0.1.89"
 base64 = "0.22.1"
 bytes = "1.10.1"
-chrono = "0.4.40"
+chrono = "0.4.42"
 clap = "4"
 clap_complete = "4"
 color-eyre = "0.6.3"
 crossterm = "0.28.1"
+ctor = "0.5.0"
 derive_more = "2"
 diffy = "0.4.2"
 dirs = "6"
 dotenvy = "0.15.7"
 env-flags = "0.1.1"
 env_logger = "0.11.5"
+escargot = "0.5"
 eventsource-stream = "0.2.3"
 futures = "0.3"
 icu_decimal = "2.0.0"
 icu_locale_core = "2.0.0"
 ignore = "0.4.23"
 image = { version = "^0.25.8", default-features = false }
+indexmap = "2.6.0"
 insta = "1.43.2"
 itertools = "0.14.0"
 landlock = "0.4.1"
@@ -94,8 +115,12 @@ maplit = "1.0.2"
 mime_guess = "2.0.5"
 multimap = "0.10.0"
 nucleo-matcher = "0.3.1"
-once_cell = "1"
 openssl-sys = "*"
+opentelemetry = "0.30.0"
+opentelemetry-appender-tracing = "0.30.0"
+opentelemetry-otlp = "0.30.0"
+opentelemetry-semantic-conventions = "0.30.0"
+opentelemetry_sdk = "0.30.0"
 os_info = "3.12.0"
 owo-colors = "4.2.0"
 path-absolutize = "3.1.1"
@@ -134,14 +159,16 @@ tokio-test = "0.4"
 tokio-util = "0.7.16"
 toml = "0.9.5"
 toml_edit = "0.23.4"
+tonic = "0.13.1"
 tracing = "0.1.41"
 tracing-appender = "0.2.3"
 tracing-subscriber = "0.3.20"
+tracing-test = "0.2.5"
 tree-sitter = "0.25.9"
 tree-sitter-bash = "0.25.0"
 ts-rs = "11"
 unicode-segmentation = "1.12.0"
-unicode-width = "0.1"
+unicode-width = "0.2"
 url = "2"
 urlencoding = "2.1"
 uuid = "1"
@@ -151,6 +178,7 @@ webbrowser = "1.0"
 which = "6"
 wildmatch = "2.5.0"
 wiremock = "0.6"
+zeroize = "1.8.1"
 
 [workspace.lints]
 rust = {}

codex-rs/README.md

@@ -4,18 +4,18 @@ We provide Codex CLI as a standalone, native executable to ensure a zero-depende
 
 ## Installing Codex
 
-Today, the easiest way to install Codex is via `npm`, though we plan to publish Codex to other package managers soon.
+Today, the easiest way to install Codex is via `npm`:
 
 ```shell
-npm i -g @openai/codex@native
+npm i -g @openai/codex
 codex
 ```
 
-You can also download a platform-specific release directly from our [GitHub Releases](https://github.com/openai/codex/releases).
+You can also install via Homebrew (`brew install codex`) or download a platform-specific release directly from our [GitHub Releases](https://github.com/openai/codex/releases).
 
 ## What's new in the Rust CLI
 
-While we are [working to close the gap between the TypeScript and Rust implementations of Codex CLI](https://github.com/openai/codex/issues/1262), note that the Rust CLI has a number of features that the TypeScript CLI does not!
+The Rust implementation is now the maintained Codex CLI and serves as the default experience. It includes a number of features that the legacy TypeScript CLI never supported.
 
 ### Config
 
@@ -25,12 +25,14 @@ Codex supports a rich set of configuration options. Note that the Rust CLI uses
 
 Codex CLI functions as an MCP client that can connect to MCP servers on startup. See the [`mcp_servers`](../docs/config.md#mcp_servers) section in the configuration documentation for details.
 
-It is still experimental, but you can also launch Codex as an MCP _server_ by running `codex mcp`. Use the [`@modelcontextprotocol/inspector`](https://github.com/modelcontextprotocol/inspector) to try it out:
+It is still experimental, but you can also launch Codex as an MCP _server_ by running `codex mcp-server`. Use the [`@modelcontextprotocol/inspector`](https://github.com/modelcontextprotocol/inspector) to try it out:
 
 ```shell
-npx @modelcontextprotocol/inspector codex mcp
+npx @modelcontextprotocol/inspector codex mcp-server
 ```
 
+Use `codex mcp` to add/list/get/remove MCP server launchers defined in `config.toml`, and `codex mcp-server` to run the MCP server directly.
+
 ### Notifications
 
 You can enable notifications by configuring a script that is run whenever the agent finishes a turn. The [notify documentation](../docs/config.md#notify) includes a detailed example that explains how to get desktop notifications via [terminal-notifier](https://github.com/julienXX/terminal-notifier) on macOS.

codex-rs/app-server/Cargo.toml

@@ -0,0 +1,51 @@
+[package]
+edition = "2024"
+name = "codex-app-server"
+version = { workspace = true }
+
+[[bin]]
+name = "codex-app-server"
+path = "src/main.rs"
+
+[lib]
+name = "codex_app_server"
+path = "src/lib.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+codex-arg0 = { workspace = true }
+codex-common = { workspace = true, features = ["cli"] }
+codex-core = { workspace = true }
+codex-file-search = { workspace = true }
+codex-login = { workspace = true }
+codex-protocol = { workspace = true }
+codex-utils-json-to-toml = { workspace = true }
+# We should only be using mcp-types for JSON-RPC types: it would be nice to
+# split this out into a separate crate at some point.
+mcp-types = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+tokio = { workspace = true, features = [
+    "io-std",
+    "macros",
+    "process",
+    "rt-multi-thread",
+    "signal",
+] }
+tracing = { workspace = true, features = ["log"] }
+tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
+uuid = { workspace = true, features = ["serde", "v7"] }
+
+[dev-dependencies]
+app_test_support = { workspace = true }
+assert_cmd = { workspace = true }
+base64 = { workspace = true }
+core_test_support = { workspace = true }
+os_info = { workspace = true }
+pretty_assertions = { workspace = true }
+tempfile = { workspace = true }
+toml = { workspace = true }
+wiremock = { workspace = true }

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

@@ -1,6 +1,6 @@
 use crate::error_code::INTERNAL_ERROR_CODE;
 use crate::error_code::INVALID_REQUEST_ERROR_CODE;
-use crate::json_to_toml::json_to_toml;
+use crate::fuzzy_file_search::run_fuzzy_file_search;
 use crate::outgoing_message::OutgoingMessageSender;
 use crate::outgoing_message::OutgoingNotification;
 use codex_core::AuthManager;
@@ -52,6 +52,8 @@ use codex_protocol::mcp_protocol::ExecArbitraryCommandResponse;
 use codex_protocol::mcp_protocol::ExecCommandApprovalParams;
 use codex_protocol::mcp_protocol::ExecCommandApprovalResponse;
 use codex_protocol::mcp_protocol::ExecOneOffCommandParams;
+use codex_protocol::mcp_protocol::FuzzyFileSearchParams;
+use codex_protocol::mcp_protocol::FuzzyFileSearchResponse;
 use codex_protocol::mcp_protocol::GetUserAgentResponse;
 use codex_protocol::mcp_protocol::GetUserSavedConfigResponse;
 use codex_protocol::mcp_protocol::GitDiffToRemoteResponse;
@@ -74,6 +76,7 @@ use codex_protocol::mcp_protocol::SendUserMessageResponse;
 use codex_protocol::mcp_protocol::SendUserTurnParams;
 use codex_protocol::mcp_protocol::SendUserTurnResponse;
 use codex_protocol::mcp_protocol::ServerNotification;
+use codex_protocol::mcp_protocol::SessionConfiguredNotification;
 use codex_protocol::mcp_protocol::SetDefaultModelParams;
 use codex_protocol::mcp_protocol::SetDefaultModelResponse;
 use codex_protocol::mcp_protocol::UserInfoResponse;
@@ -82,12 +85,15 @@ use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::InputMessageKind;
 use codex_protocol::protocol::USER_MESSAGE_BEGIN;
+use codex_utils_json_to_toml::json_to_toml;
 use mcp_types::JSONRPCErrorError;
 use mcp_types::RequestId;
 use std::collections::HashMap;
 use std::ffi::OsStr;
 use std::path::PathBuf;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::time::Duration;
 use tokio::select;
 use tokio::sync::Mutex;
@@ -122,6 +128,7 @@ pub(crate) struct CodexMessageProcessor {
     active_login: Arc<Mutex<Option<ActiveLogin>>>,
     // Queue of pending interrupt requests per conversation. We reply when TurnAborted arrives.
     pending_interrupts: Arc<Mutex<HashMap<ConversationId, Vec<RequestId>>>>,
+    pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
 }
 
 impl CodexMessageProcessor {
@@ -141,11 +148,15 @@ impl CodexMessageProcessor {
             conversation_listeners: HashMap::new(),
             active_login: Arc::new(Mutex::new(None)),
             pending_interrupts: Arc::new(Mutex::new(HashMap::new())),
+            pending_fuzzy_searches: Arc::new(Mutex::new(HashMap::new())),
         }
     }
 
     pub async fn process_request(&mut self, request: ClientRequest) {
         match request {
+            ClientRequest::Initialize { .. } => {
+                panic!("Initialize should be handled in MessageProcessor");
+            }
             ClientRequest::NewConversation { request_id, params } => {
                 // Do not tokio::spawn() to process new_conversation()
                 // asynchronously because we need to ensure the conversation is
@@ -206,6 +217,9 @@ impl CodexMessageProcessor {
             ClientRequest::UserInfo { request_id } => {
                 self.get_user_info(request_id).await;
             }
+            ClientRequest::FuzzyFileSearch { request_id, params } => {
+                self.fuzzy_file_search(request_id, params).await;
+            }
             ClientRequest::ExecOneOffCommand { request_id, params } => {
                 self.exec_one_off_command(request_id, params).await;
             }
@@ -752,11 +766,19 @@ impl CodexMessageProcessor {
                 session_configured,
                 ..
             }) => {
-                let event = Event {
-                    id: "".to_string(),
-                    msg: EventMsg::SessionConfigured(session_configured.clone()),
-                };
-                self.outgoing.send_event_as_notification(&event, None).await;
+                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()
                         .filter(|event| {
@@ -1167,6 +1189,46 @@ impl CodexMessageProcessor {
             }
         }
     }
+
+    async fn fuzzy_file_search(&mut self, request_id: RequestId, params: FuzzyFileSearchParams) {
+        let FuzzyFileSearchParams {
+            query,
+            roots,
+            cancellation_token,
+        } = params;
+
+        let cancel_flag = match cancellation_token.clone() {
+            Some(token) => {
+                let mut pending_fuzzy_searches = self.pending_fuzzy_searches.lock().await;
+                // if a cancellation_token is provided and a pending_request exists for
+                // that token, cancel it
+                if let Some(existing) = pending_fuzzy_searches.get(&token) {
+                    existing.store(true, Ordering::Relaxed);
+                }
+                let flag = Arc::new(AtomicBool::new(false));
+                pending_fuzzy_searches.insert(token.clone(), flag.clone());
+                flag
+            }
+            None => Arc::new(AtomicBool::new(false)),
+        };
+
+        let results = match query.as_str() {
+            "" => vec![],
+            _ => run_fuzzy_file_search(query, roots, cancel_flag.clone()).await,
+        };
+
+        if let Some(token) = cancellation_token {
+            let mut pending_fuzzy_searches = self.pending_fuzzy_searches.lock().await;
+            if let Some(current_flag) = pending_fuzzy_searches.get(&token)
+                && Arc::ptr_eq(current_flag, &cancel_flag)
+            {
+                pending_fuzzy_searches.remove(&token);
+            }
+        }
+
+        let response = FuzzyFileSearchResponse { files: results };
+        self.outgoing.send_response(request_id, response).await;
+    }
 }
 
 async fn apply_bespoke_event_handling(

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

@@ -0,0 +1,2 @@
+pub(crate) const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
+pub(crate) const INTERNAL_ERROR_CODE: i64 = -32603;

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

@@ -0,0 +1,84 @@
+use std::num::NonZero;
+use std::num::NonZeroUsize;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+
+use codex_file_search as file_search;
+use codex_protocol::mcp_protocol::FuzzyFileSearchResult;
+use tokio::task::JoinSet;
+use tracing::warn;
+
+const LIMIT_PER_ROOT: usize = 50;
+const MAX_THREADS: usize = 12;
+const COMPUTE_INDICES: bool = true;
+
+pub(crate) async fn run_fuzzy_file_search(
+    query: String,
+    roots: Vec<String>,
+    cancellation_flag: Arc<AtomicBool>,
+) -> Vec<FuzzyFileSearchResult> {
+    #[expect(clippy::expect_used)]
+    let limit_per_root =
+        NonZero::new(LIMIT_PER_ROOT).expect("LIMIT_PER_ROOT should be a valid non-zero usize");
+
+    let cores = std::thread::available_parallelism()
+        .map(std::num::NonZero::get)
+        .unwrap_or(1);
+    let threads = cores.min(MAX_THREADS);
+    let threads_per_root = (threads / roots.len()).max(1);
+    let threads = NonZero::new(threads_per_root).unwrap_or(NonZeroUsize::MIN);
+
+    let mut files: Vec<FuzzyFileSearchResult> = Vec::new();
+    let mut join_set = JoinSet::new();
+
+    for root in roots {
+        let search_dir = PathBuf::from(&root);
+        let query = query.clone();
+        let cancel_flag = cancellation_flag.clone();
+        join_set.spawn_blocking(move || {
+            match file_search::run(
+                query.as_str(),
+                limit_per_root,
+                &search_dir,
+                Vec::new(),
+                threads,
+                cancel_flag,
+                COMPUTE_INDICES,
+            ) {
+                Ok(res) => Ok((root, res)),
+                Err(err) => Err((root, err)),
+            }
+        });
+    }
+
+    while let Some(res) = join_set.join_next().await {
+        match res {
+            Ok(Ok((root, res))) => {
+                for m in res.matches {
+                    let result = FuzzyFileSearchResult {
+                        root: root.clone(),
+                        path: m.path,
+                        score: m.score,
+                        indices: m.indices,
+                    };
+                    files.push(result);
+                }
+            }
+            Ok(Err((root, err))) => {
+                warn!("fuzzy-file-search in dir '{root}' failed: {err}");
+            }
+            Err(err) => {
+                warn!("fuzzy-file-search join_next failed: {err}");
+            }
+        }
+    }
+
+    files.sort_by(file_search::cmp_by_score_desc_then_path_asc::<
+        FuzzyFileSearchResult,
+        _,
+        _,
+    >(|f| f.score, |f| f.path.as_str()));
+
+    files
+}

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

@@ -0,0 +1,139 @@
+#![deny(clippy::print_stdout, clippy::print_stderr)]
+
+use std::io::ErrorKind;
+use std::io::Result as IoResult;
+use std::path::PathBuf;
+
+use codex_common::CliConfigOverrides;
+use codex_core::config::Config;
+use codex_core::config::ConfigOverrides;
+
+use mcp_types::JSONRPCMessage;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::io::{self};
+use tokio::sync::mpsc;
+use tracing::debug;
+use tracing::error;
+use tracing::info;
+use tracing_subscriber::EnvFilter;
+
+use crate::message_processor::MessageProcessor;
+use crate::outgoing_message::OutgoingMessage;
+use crate::outgoing_message::OutgoingMessageSender;
+
+mod codex_message_processor;
+mod error_code;
+mod fuzzy_file_search;
+mod message_processor;
+mod outgoing_message;
+
+/// Size of the bounded channels used to communicate between tasks. The value
+/// is a balance between throughput and memory usage – 128 messages should be
+/// plenty for an interactive CLI.
+const CHANNEL_CAPACITY: usize = 128;
+
+pub async fn run_main(
+    codex_linux_sandbox_exe: Option<PathBuf>,
+    cli_config_overrides: CliConfigOverrides,
+) -> IoResult<()> {
+    // Install a simple subscriber so `tracing` output is visible.  Users can
+    // control the log level with `RUST_LOG`.
+    tracing_subscriber::fmt()
+        .with_writer(std::io::stderr)
+        .with_env_filter(EnvFilter::from_default_env())
+        .init();
+
+    // Set up channels.
+    let (incoming_tx, mut incoming_rx) = mpsc::channel::<JSONRPCMessage>(CHANNEL_CAPACITY);
+    let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<OutgoingMessage>();
+
+    // Task: read from stdin, push to `incoming_tx`.
+    let stdin_reader_handle = tokio::spawn({
+        async move {
+            let stdin = io::stdin();
+            let reader = BufReader::new(stdin);
+            let mut lines = reader.lines();
+
+            while let Some(line) = lines.next_line().await.unwrap_or_default() {
+                match serde_json::from_str::<JSONRPCMessage>(&line) {
+                    Ok(msg) => {
+                        if incoming_tx.send(msg).await.is_err() {
+                            // Receiver gone – nothing left to do.
+                            break;
+                        }
+                    }
+                    Err(e) => error!("Failed to deserialize JSONRPCMessage: {e}"),
+                }
+            }
+
+            debug!("stdin reader finished (EOF)");
+        }
+    });
+
+    // Parse CLI overrides once and derive the base Config eagerly so later
+    // components do not need to work with raw TOML values.
+    let cli_kv_overrides = cli_config_overrides.parse_overrides().map_err(|e| {
+        std::io::Error::new(
+            ErrorKind::InvalidInput,
+            format!("error parsing -c overrides: {e}"),
+        )
+    })?;
+    let config = Config::load_with_cli_overrides(cli_kv_overrides, ConfigOverrides::default())
+        .map_err(|e| {
+            std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
+        })?;
+
+    // Task: process incoming messages.
+    let processor_handle = tokio::spawn({
+        let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
+        let mut processor = MessageProcessor::new(
+            outgoing_message_sender,
+            codex_linux_sandbox_exe,
+            std::sync::Arc::new(config),
+        );
+        async move {
+            while let Some(msg) = incoming_rx.recv().await {
+                match msg {
+                    JSONRPCMessage::Request(r) => processor.process_request(r).await,
+                    JSONRPCMessage::Response(r) => processor.process_response(r).await,
+                    JSONRPCMessage::Notification(n) => processor.process_notification(n).await,
+                    JSONRPCMessage::Error(e) => processor.process_error(e),
+                }
+            }
+
+            info!("processor task exited (channel closed)");
+        }
+    });
+
+    // Task: write outgoing messages to stdout.
+    let stdout_writer_handle = tokio::spawn(async move {
+        let mut stdout = io::stdout();
+        while let Some(outgoing_message) = outgoing_rx.recv().await {
+            let msg: JSONRPCMessage = outgoing_message.into();
+            match serde_json::to_string(&msg) {
+                Ok(json) => {
+                    if let Err(e) = stdout.write_all(json.as_bytes()).await {
+                        error!("Failed to write to stdout: {e}");
+                        break;
+                    }
+                    if let Err(e) = stdout.write_all(b"\n").await {
+                        error!("Failed to write newline to stdout: {e}");
+                        break;
+                    }
+                }
+                Err(e) => error!("Failed to serialize JSONRPCMessage: {e}"),
+            }
+        }
+
+        info!("stdout writer exited (channel closed)");
+    });
+
+    // Wait for all tasks to finish.  The typical exit path is the stdin reader
+    // hitting EOF which, once it drops `incoming_tx`, propagates shutdown to
+    // the processor and then to the stdout task.
+    let _ = tokio::join!(stdin_reader_handle, processor_handle, stdout_writer_handle);
+
+    Ok(())
+}

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

@@ -0,0 +1,10 @@
+use codex_app_server::run_main;
+use codex_arg0::arg0_dispatch_or_else;
+use codex_common::CliConfigOverrides;
+
+fn main() -> anyhow::Result<()> {
+    arg0_dispatch_or_else(|codex_linux_sandbox_exe| async move {
+        run_main(codex_linux_sandbox_exe, CliConfigOverrides::default()).await?;
+        Ok(())
+    })
+}

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

@@ -0,0 +1,133 @@
+use std::path::PathBuf;
+
+use crate::codex_message_processor::CodexMessageProcessor;
+use crate::error_code::INVALID_REQUEST_ERROR_CODE;
+use crate::outgoing_message::OutgoingMessageSender;
+use codex_protocol::mcp_protocol::ClientInfo;
+use codex_protocol::mcp_protocol::ClientRequest;
+use codex_protocol::mcp_protocol::InitializeResponse;
+
+use codex_core::AuthManager;
+use codex_core::ConversationManager;
+use codex_core::config::Config;
+use codex_core::default_client::USER_AGENT_SUFFIX;
+use codex_core::default_client::get_codex_user_agent;
+use mcp_types::JSONRPCError;
+use mcp_types::JSONRPCErrorError;
+use mcp_types::JSONRPCNotification;
+use mcp_types::JSONRPCRequest;
+use mcp_types::JSONRPCResponse;
+use std::sync::Arc;
+
+pub(crate) struct MessageProcessor {
+    outgoing: Arc<OutgoingMessageSender>,
+    codex_message_processor: CodexMessageProcessor,
+    initialized: bool,
+}
+
+impl MessageProcessor {
+    /// Create a new `MessageProcessor`, retaining a handle to the outgoing
+    /// `Sender` so handlers can enqueue messages to be written to stdout.
+    pub(crate) fn new(
+        outgoing: OutgoingMessageSender,
+        codex_linux_sandbox_exe: Option<PathBuf>,
+        config: Arc<Config>,
+    ) -> Self {
+        let outgoing = Arc::new(outgoing);
+        let auth_manager = AuthManager::shared(config.codex_home.clone());
+        let conversation_manager = Arc::new(ConversationManager::new(auth_manager.clone()));
+        let codex_message_processor = CodexMessageProcessor::new(
+            auth_manager,
+            conversation_manager,
+            outgoing.clone(),
+            codex_linux_sandbox_exe,
+            config,
+        );
+
+        Self {
+            outgoing,
+            codex_message_processor,
+            initialized: false,
+        }
+    }
+
+    pub(crate) async fn process_request(&mut self, request: JSONRPCRequest) {
+        let request_id = request.id.clone();
+        if let Ok(request_json) = serde_json::to_value(request)
+            && let Ok(codex_request) = serde_json::from_value::<ClientRequest>(request_json)
+        {
+            match codex_request {
+                // Handle Initialize internally so CodexMessageProcessor does not have to concern
+                // itself with the `initialized` bool.
+                ClientRequest::Initialize { request_id, params } => {
+                    if self.initialized {
+                        let error = JSONRPCErrorError {
+                            code: INVALID_REQUEST_ERROR_CODE,
+                            message: "Already initialized".to_string(),
+                            data: None,
+                        };
+                        self.outgoing.send_error(request_id, error).await;
+                        return;
+                    } else {
+                        let ClientInfo {
+                            name,
+                            title: _title,
+                            version,
+                        } = params.client_info;
+                        let user_agent_suffix = format!("{name}; {version}");
+                        if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
+                            *suffix = Some(user_agent_suffix);
+                        }
+
+                        let user_agent = get_codex_user_agent();
+                        let response = InitializeResponse { user_agent };
+                        self.outgoing.send_response(request_id, response).await;
+
+                        self.initialized = true;
+                        return;
+                    }
+                }
+                _ => {
+                    if !self.initialized {
+                        let error = JSONRPCErrorError {
+                            code: INVALID_REQUEST_ERROR_CODE,
+                            message: "Not initialized".to_string(),
+                            data: None,
+                        };
+                        self.outgoing.send_error(request_id, error).await;
+                        return;
+                    }
+                }
+            }
+
+            self.codex_message_processor
+                .process_request(codex_request)
+                .await;
+        } else {
+            let error = JSONRPCErrorError {
+                code: INVALID_REQUEST_ERROR_CODE,
+                message: "Invalid request".to_string(),
+                data: None,
+            };
+            self.outgoing.send_error(request_id, error).await;
+        }
+    }
+
+    pub(crate) async fn process_notification(&self, notification: JSONRPCNotification) {
+        // Currently, we do not expect to receive any notifications from the
+        // client, so we just log them.
+        tracing::info!("<- notification: {:?}", notification);
+    }
+
+    /// Handle a standalone JSON-RPC response originating from the peer.
+    pub(crate) async fn process_response(&mut self, response: JSONRPCResponse) {
+        tracing::info!("<- response: {:?}", response);
+        let JSONRPCResponse { id, result, .. } = response;
+        self.outgoing.notify_client_response(id, result).await
+    }
+
+    /// Handle an error object received from the peer.
+    pub(crate) fn process_error(&mut self, err: JSONRPCError) {
+        tracing::error!("<- error: {:?}", err);
+    }
+}

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

@@ -0,0 +1,239 @@
+use std::collections::HashMap;
+use std::sync::atomic::AtomicI64;
+use std::sync::atomic::Ordering;
+
+use codex_protocol::mcp_protocol::ServerNotification;
+use mcp_types::JSONRPC_VERSION;
+use mcp_types::JSONRPCError;
+use mcp_types::JSONRPCErrorError;
+use mcp_types::JSONRPCMessage;
+use mcp_types::JSONRPCNotification;
+use mcp_types::JSONRPCRequest;
+use mcp_types::JSONRPCResponse;
+use mcp_types::RequestId;
+use mcp_types::Result;
+use serde::Serialize;
+use tokio::sync::Mutex;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tracing::warn;
+
+use crate::error_code::INTERNAL_ERROR_CODE;
+
+/// Sends messages to the client and manages request callbacks.
+pub(crate) struct OutgoingMessageSender {
+    next_request_id: AtomicI64,
+    sender: mpsc::UnboundedSender<OutgoingMessage>,
+    request_id_to_callback: Mutex<HashMap<RequestId, oneshot::Sender<Result>>>,
+}
+
+impl OutgoingMessageSender {
+    pub(crate) fn new(sender: mpsc::UnboundedSender<OutgoingMessage>) -> Self {
+        Self {
+            next_request_id: AtomicI64::new(0),
+            sender,
+            request_id_to_callback: Mutex::new(HashMap::new()),
+        }
+    }
+
+    pub(crate) async fn send_request(
+        &self,
+        method: &str,
+        params: Option<serde_json::Value>,
+    ) -> oneshot::Receiver<Result> {
+        let id = RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::Relaxed));
+        let outgoing_message_id = id.clone();
+        let (tx_approve, rx_approve) = oneshot::channel();
+        {
+            let mut request_id_to_callback = self.request_id_to_callback.lock().await;
+            request_id_to_callback.insert(id, tx_approve);
+        }
+
+        let outgoing_message = OutgoingMessage::Request(OutgoingRequest {
+            id: outgoing_message_id,
+            method: method.to_string(),
+            params,
+        });
+        let _ = self.sender.send(outgoing_message);
+        rx_approve
+    }
+
+    pub(crate) async fn notify_client_response(&self, id: RequestId, result: Result) {
+        let entry = {
+            let mut request_id_to_callback = self.request_id_to_callback.lock().await;
+            request_id_to_callback.remove_entry(&id)
+        };
+
+        match entry {
+            Some((id, sender)) => {
+                if let Err(err) = sender.send(result) {
+                    warn!("could not notify callback for {id:?} due to: {err:?}");
+                }
+            }
+            None => {
+                warn!("could not find callback for {id:?}");
+            }
+        }
+    }
+
+    pub(crate) async fn send_response<T: Serialize>(&self, id: RequestId, response: T) {
+        match serde_json::to_value(response) {
+            Ok(result) => {
+                let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result });
+                let _ = self.sender.send(outgoing_message);
+            }
+            Err(err) => {
+                self.send_error(
+                    id,
+                    JSONRPCErrorError {
+                        code: INTERNAL_ERROR_CODE,
+                        message: format!("failed to serialize response: {err}"),
+                        data: None,
+                    },
+                )
+                .await;
+            }
+        }
+    }
+
+    pub(crate) async fn send_server_notification(&self, notification: ServerNotification) {
+        let _ = self
+            .sender
+            .send(OutgoingMessage::AppServerNotification(notification));
+    }
+
+    /// All notifications should be migrated to [`ServerNotification`] and
+    /// [`OutgoingMessage::Notification`] should be removed.
+    pub(crate) async fn send_notification(&self, notification: OutgoingNotification) {
+        let outgoing_message = OutgoingMessage::Notification(notification);
+        let _ = self.sender.send(outgoing_message);
+    }
+
+    pub(crate) async fn send_error(&self, id: RequestId, error: JSONRPCErrorError) {
+        let outgoing_message = OutgoingMessage::Error(OutgoingError { id, error });
+        let _ = self.sender.send(outgoing_message);
+    }
+}
+
+/// Outgoing message from the server to the client.
+pub(crate) enum OutgoingMessage {
+    Request(OutgoingRequest),
+    Notification(OutgoingNotification),
+    /// AppServerNotification is specific to the case where this is run as an
+    /// "app server" as opposed to an MCP server.
+    AppServerNotification(ServerNotification),
+    Response(OutgoingResponse),
+    Error(OutgoingError),
+}
+
+impl From<OutgoingMessage> for JSONRPCMessage {
+    fn from(val: OutgoingMessage) -> Self {
+        use OutgoingMessage::*;
+        match val {
+            Request(OutgoingRequest { id, method, params }) => {
+                JSONRPCMessage::Request(JSONRPCRequest {
+                    jsonrpc: JSONRPC_VERSION.into(),
+                    id,
+                    method,
+                    params,
+                })
+            }
+            Notification(OutgoingNotification { method, params }) => {
+                JSONRPCMessage::Notification(JSONRPCNotification {
+                    jsonrpc: JSONRPC_VERSION.into(),
+                    method,
+                    params,
+                })
+            }
+            AppServerNotification(notification) => {
+                let method = notification.to_string();
+                let params = match notification.to_params() {
+                    Ok(params) => Some(params),
+                    Err(err) => {
+                        warn!("failed to serialize notification params: {err}");
+                        None
+                    }
+                };
+                JSONRPCMessage::Notification(JSONRPCNotification {
+                    jsonrpc: JSONRPC_VERSION.into(),
+                    method,
+                    params,
+                })
+            }
+            Response(OutgoingResponse { id, result }) => {
+                JSONRPCMessage::Response(JSONRPCResponse {
+                    jsonrpc: JSONRPC_VERSION.into(),
+                    id,
+                    result,
+                })
+            }
+            Error(OutgoingError { id, error }) => JSONRPCMessage::Error(JSONRPCError {
+                jsonrpc: JSONRPC_VERSION.into(),
+                id,
+                error,
+            }),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize)]
+pub(crate) struct OutgoingRequest {
+    pub id: RequestId,
+    pub method: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub params: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize)]
+pub(crate) struct OutgoingNotification {
+    pub method: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub params: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize)]
+pub(crate) struct OutgoingResponse {
+    pub id: RequestId,
+    pub result: Result,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize)]
+pub(crate) struct OutgoingError {
+    pub error: JSONRPCErrorError,
+    pub id: RequestId,
+}
+
+#[cfg(test)]
+mod tests {
+    use codex_protocol::mcp_protocol::LoginChatGptCompleteNotification;
+    use pretty_assertions::assert_eq;
+    use serde_json::json;
+    use uuid::Uuid;
+
+    use super::*;
+
+    #[test]
+    fn verify_server_notification_serialization() {
+        let notification =
+            ServerNotification::LoginChatGptComplete(LoginChatGptCompleteNotification {
+                login_id: Uuid::nil(),
+                success: true,
+                error: None,
+            });
+
+        let jsonrpc_notification: JSONRPCMessage =
+            OutgoingMessage::AppServerNotification(notification).into();
+        assert_eq!(
+            JSONRPCMessage::Notification(JSONRPCNotification {
+                jsonrpc: "2.0".into(),
+                method: "loginChatGptComplete".into(),
+                params: Some(json!({
+                    "loginId": Uuid::nil(),
+                    "success": true,
+                })),
+            }),
+            jsonrpc_notification,
+            "ensure the strum macros serialize the method field correctly"
+        );
+    }
+}

codex-rs/app-server/tests/all.rs

@@ -0,0 +1,3 @@
+// Single integration test binary that aggregates all test modules.
+// The submodules live in `tests/suite/`.
+mod suite;

codex-rs/app-server/tests/common/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+edition = "2024"
+name = "app_test_support"
+version = { workspace = true }
+
+[lib]
+path = "lib.rs"
+
+[dependencies]
+anyhow = { workspace = true }
+assert_cmd = { workspace = true }
+codex-protocol = { workspace = true }
+mcp-types = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true, features = [
+    "io-std",
+    "macros",
+    "process",
+    "rt-multi-thread",
+] }
+wiremock = { workspace = true }

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

@@ -0,0 +1,17 @@
+mod mcp_process;
+mod mock_model_server;
+mod responses;
+
+pub use mcp_process::McpProcess;
+use mcp_types::JSONRPCResponse;
+pub use mock_model_server::create_mock_chat_completions_server;
+pub use responses::create_apply_patch_sse_response;
+pub use responses::create_final_assistant_message_sse_response;
+pub use responses::create_shell_sse_response;
+use serde::de::DeserializeOwned;
+
+pub fn to_response<T: DeserializeOwned>(response: JSONRPCResponse) -> anyhow::Result<T> {
+    let value = serde_json::to_value(response.result)?;
+    let codex_response = serde_json::from_value(value)?;
+    Ok(codex_response)
+}

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

@@ -0,0 +1,477 @@
+use std::path::Path;
+use std::process::Stdio;
+use std::sync::atomic::AtomicI64;
+use std::sync::atomic::Ordering;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::process::Child;
+use tokio::process::ChildStdin;
+use tokio::process::ChildStdout;
+
+use anyhow::Context;
+use assert_cmd::prelude::*;
+use codex_protocol::mcp_protocol::AddConversationListenerParams;
+use codex_protocol::mcp_protocol::ArchiveConversationParams;
+use codex_protocol::mcp_protocol::CancelLoginChatGptParams;
+use codex_protocol::mcp_protocol::ClientInfo;
+use codex_protocol::mcp_protocol::ClientNotification;
+use codex_protocol::mcp_protocol::GetAuthStatusParams;
+use codex_protocol::mcp_protocol::InitializeParams;
+use codex_protocol::mcp_protocol::InterruptConversationParams;
+use codex_protocol::mcp_protocol::ListConversationsParams;
+use codex_protocol::mcp_protocol::LoginApiKeyParams;
+use codex_protocol::mcp_protocol::NewConversationParams;
+use codex_protocol::mcp_protocol::RemoveConversationListenerParams;
+use codex_protocol::mcp_protocol::ResumeConversationParams;
+use codex_protocol::mcp_protocol::SendUserMessageParams;
+use codex_protocol::mcp_protocol::SendUserTurnParams;
+use codex_protocol::mcp_protocol::SetDefaultModelParams;
+
+use mcp_types::JSONRPC_VERSION;
+use mcp_types::JSONRPCMessage;
+use mcp_types::JSONRPCNotification;
+use mcp_types::JSONRPCRequest;
+use mcp_types::JSONRPCResponse;
+use mcp_types::RequestId;
+use std::process::Command as StdCommand;
+use tokio::process::Command;
+
+pub struct McpProcess {
+    next_request_id: AtomicI64,
+    /// Retain this child process until the client is dropped. The Tokio runtime
+    /// will make a "best effort" to reap the process after it exits, but it is
+    /// not a guarantee. See the `kill_on_drop` documentation for details.
+    #[allow(dead_code)]
+    process: Child,
+    stdin: ChildStdin,
+    stdout: BufReader<ChildStdout>,
+}
+
+impl McpProcess {
+    pub async fn new(codex_home: &Path) -> anyhow::Result<Self> {
+        Self::new_with_env(codex_home, &[]).await
+    }
+
+    /// Creates a new MCP process, allowing tests to override or remove
+    /// specific environment variables for the child process only.
+    ///
+    /// Pass a tuple of (key, Some(value)) to set/override, or (key, None) to
+    /// remove a variable from the child's environment.
+    pub async fn new_with_env(
+        codex_home: &Path,
+        env_overrides: &[(&str, Option<&str>)],
+    ) -> anyhow::Result<Self> {
+        // Use assert_cmd to locate the binary path and then switch to tokio::process::Command
+        let std_cmd = StdCommand::cargo_bin("codex-app-server")
+            .context("should find binary for codex-mcp-server")?;
+
+        let program = std_cmd.get_program().to_owned();
+
+        let mut cmd = Command::new(program);
+
+        cmd.stdin(Stdio::piped());
+        cmd.stdout(Stdio::piped());
+        cmd.stderr(Stdio::piped());
+        cmd.env("CODEX_HOME", codex_home);
+        cmd.env("RUST_LOG", "debug");
+
+        for (k, v) in env_overrides {
+            match v {
+                Some(val) => {
+                    cmd.env(k, val);
+                }
+                None => {
+                    cmd.env_remove(k);
+                }
+            }
+        }
+
+        let mut process = cmd
+            .kill_on_drop(true)
+            .spawn()
+            .context("codex-mcp-server proc should start")?;
+        let stdin = process
+            .stdin
+            .take()
+            .ok_or_else(|| anyhow::format_err!("mcp should have stdin fd"))?;
+        let stdout = process
+            .stdout
+            .take()
+            .ok_or_else(|| anyhow::format_err!("mcp should have stdout fd"))?;
+        let stdout = BufReader::new(stdout);
+
+        // Forward child's stderr to our stderr so failures are visible even
+        // when stdout/stderr are captured by the test harness.
+        if let Some(stderr) = process.stderr.take() {
+            let mut stderr_reader = BufReader::new(stderr).lines();
+            tokio::spawn(async move {
+                while let Ok(Some(line)) = stderr_reader.next_line().await {
+                    eprintln!("[mcp stderr] {line}");
+                }
+            });
+        }
+        Ok(Self {
+            next_request_id: AtomicI64::new(0),
+            process,
+            stdin,
+            stdout,
+        })
+    }
+
+    /// 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(),
+                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 {
+            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(())
+    }
+
+    /// Send a `newConversation` JSON-RPC request.
+    pub async fn send_new_conversation_request(
+        &mut self,
+        params: NewConversationParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("newConversation", params).await
+    }
+
+    /// Send an `archiveConversation` JSON-RPC request.
+    pub async fn send_archive_conversation_request(
+        &mut self,
+        params: ArchiveConversationParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("archiveConversation", params).await
+    }
+
+    /// Send an `addConversationListener` JSON-RPC request.
+    pub async fn send_add_conversation_listener_request(
+        &mut self,
+        params: AddConversationListenerParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("addConversationListener", params).await
+    }
+
+    /// Send a `sendUserMessage` JSON-RPC request with a single text item.
+    pub async fn send_send_user_message_request(
+        &mut self,
+        params: SendUserMessageParams,
+    ) -> anyhow::Result<i64> {
+        // Wire format expects variants in camelCase; text item uses external tagging.
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("sendUserMessage", params).await
+    }
+
+    /// Send a `removeConversationListener` JSON-RPC request.
+    pub async fn send_remove_conversation_listener_request(
+        &mut self,
+        params: RemoveConversationListenerParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("removeConversationListener", params)
+            .await
+    }
+
+    /// Send a `sendUserTurn` JSON-RPC request.
+    pub async fn send_send_user_turn_request(
+        &mut self,
+        params: SendUserTurnParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("sendUserTurn", params).await
+    }
+
+    /// Send a `interruptConversation` JSON-RPC request.
+    pub async fn send_interrupt_conversation_request(
+        &mut self,
+        params: InterruptConversationParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("interruptConversation", params).await
+    }
+
+    /// Send a `getAuthStatus` JSON-RPC request.
+    pub async fn send_get_auth_status_request(
+        &mut self,
+        params: GetAuthStatusParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("getAuthStatus", params).await
+    }
+
+    /// Send a `getUserSavedConfig` JSON-RPC request.
+    pub async fn send_get_user_saved_config_request(&mut self) -> anyhow::Result<i64> {
+        self.send_request("getUserSavedConfig", None).await
+    }
+
+    /// Send a `getUserAgent` JSON-RPC request.
+    pub async fn send_get_user_agent_request(&mut self) -> anyhow::Result<i64> {
+        self.send_request("getUserAgent", None).await
+    }
+
+    /// Send a `userInfo` JSON-RPC request.
+    pub async fn send_user_info_request(&mut self) -> anyhow::Result<i64> {
+        self.send_request("userInfo", None).await
+    }
+
+    /// Send a `setDefaultModel` JSON-RPC request.
+    pub async fn send_set_default_model_request(
+        &mut self,
+        params: SetDefaultModelParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("setDefaultModel", params).await
+    }
+
+    /// Send a `listConversations` JSON-RPC request.
+    pub async fn send_list_conversations_request(
+        &mut self,
+        params: ListConversationsParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("listConversations", params).await
+    }
+
+    /// Send a `resumeConversation` JSON-RPC request.
+    pub async fn send_resume_conversation_request(
+        &mut self,
+        params: ResumeConversationParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("resumeConversation", params).await
+    }
+
+    /// Send a `loginApiKey` JSON-RPC request.
+    pub async fn send_login_api_key_request(
+        &mut self,
+        params: LoginApiKeyParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("loginApiKey", params).await
+    }
+
+    /// Send a `loginChatGpt` JSON-RPC request.
+    pub async fn send_login_chat_gpt_request(&mut self) -> anyhow::Result<i64> {
+        self.send_request("loginChatGpt", None).await
+    }
+
+    /// Send a `cancelLoginChatGpt` JSON-RPC request.
+    pub async fn send_cancel_login_chat_gpt_request(
+        &mut self,
+        params: CancelLoginChatGptParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("cancelLoginChatGpt", params).await
+    }
+
+    /// Send a `logoutChatGpt` JSON-RPC request.
+    pub async fn send_logout_chat_gpt_request(&mut self) -> anyhow::Result<i64> {
+        self.send_request("logoutChatGpt", None).await
+    }
+
+    /// Send a `fuzzyFileSearch` JSON-RPC request.
+    pub async fn send_fuzzy_file_search_request(
+        &mut self,
+        query: &str,
+        roots: Vec<String>,
+        cancellation_token: Option<String>,
+    ) -> anyhow::Result<i64> {
+        let mut params = serde_json::json!({
+            "query": query,
+            "roots": roots,
+        });
+        if let Some(token) = cancellation_token {
+            params["cancellationToken"] = serde_json::json!(token);
+        }
+        self.send_request("fuzzyFileSearch", Some(params)).await
+    }
+
+    async fn send_request(
+        &mut self,
+        method: &str,
+        params: Option<serde_json::Value>,
+    ) -> anyhow::Result<i64> {
+        let request_id = self.next_request_id.fetch_add(1, Ordering::Relaxed);
+
+        let message = JSONRPCMessage::Request(JSONRPCRequest {
+            jsonrpc: JSONRPC_VERSION.into(),
+            id: RequestId::Integer(request_id),
+            method: method.to_string(),
+            params,
+        });
+        self.send_jsonrpc_message(message).await?;
+        Ok(request_id)
+    }
+
+    pub async fn send_response(
+        &mut self,
+        id: RequestId,
+        result: serde_json::Value,
+    ) -> anyhow::Result<()> {
+        self.send_jsonrpc_message(JSONRPCMessage::Response(JSONRPCResponse {
+            jsonrpc: JSONRPC_VERSION.into(),
+            id,
+            result,
+        }))
+        .await
+    }
+
+    pub async fn send_notification(
+        &mut self,
+        notification: ClientNotification,
+    ) -> anyhow::Result<()> {
+        let value = serde_json::to_value(notification)?;
+        self.send_jsonrpc_message(JSONRPCMessage::Notification(JSONRPCNotification {
+            jsonrpc: JSONRPC_VERSION.into(),
+            method: value
+                .get("method")
+                .and_then(|m| m.as_str())
+                .ok_or_else(|| anyhow::format_err!("notification missing method field"))?
+                .to_string(),
+            params: value.get("params").cloned(),
+        }))
+        .await
+    }
+
+    async fn send_jsonrpc_message(&mut self, message: JSONRPCMessage) -> anyhow::Result<()> {
+        eprintln!("writing message to stdin: {message:?}");
+        let payload = serde_json::to_string(&message)?;
+        self.stdin.write_all(payload.as_bytes()).await?;
+        self.stdin.write_all(b"\n").await?;
+        self.stdin.flush().await?;
+        Ok(())
+    }
+
+    async fn read_jsonrpc_message(&mut self) -> anyhow::Result<JSONRPCMessage> {
+        let mut line = String::new();
+        self.stdout.read_line(&mut line).await?;
+        let message = serde_json::from_str::<JSONRPCMessage>(&line)?;
+        eprintln!("read message from stdout: {message:?}");
+        Ok(message)
+    }
+
+    pub async fn read_stream_until_request_message(&mut self) -> anyhow::Result<JSONRPCRequest> {
+        eprintln!("in read_stream_until_request_message()");
+
+        loop {
+            let message = self.read_jsonrpc_message().await?;
+
+            match message {
+                JSONRPCMessage::Notification(_) => {
+                    eprintln!("notification: {message:?}");
+                }
+                JSONRPCMessage::Request(jsonrpc_request) => {
+                    return Ok(jsonrpc_request);
+                }
+                JSONRPCMessage::Error(_) => {
+                    anyhow::bail!("unexpected JSONRPCMessage::Error: {message:?}");
+                }
+                JSONRPCMessage::Response(_) => {
+                    anyhow::bail!("unexpected JSONRPCMessage::Response: {message:?}");
+                }
+            }
+        }
+    }
+
+    pub async fn read_stream_until_response_message(
+        &mut self,
+        request_id: RequestId,
+    ) -> anyhow::Result<JSONRPCResponse> {
+        eprintln!("in read_stream_until_response_message({request_id:?})");
+
+        loop {
+            let message = self.read_jsonrpc_message().await?;
+            match message {
+                JSONRPCMessage::Notification(_) => {
+                    eprintln!("notification: {message:?}");
+                }
+                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);
+                    }
+                }
+            }
+        }
+    }
+
+    pub async fn read_stream_until_error_message(
+        &mut self,
+        request_id: RequestId,
+    ) -> anyhow::Result<mcp_types::JSONRPCError> {
+        loop {
+            let message = self.read_jsonrpc_message().await?;
+            match message {
+                JSONRPCMessage::Notification(_) => {
+                    eprintln!("notification: {message:?}");
+                }
+                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);
+                    }
+                }
+            }
+        }
+    }
+
+    pub async fn read_stream_until_notification_message(
+        &mut self,
+        method: &str,
+    ) -> anyhow::Result<JSONRPCNotification> {
+        eprintln!("in read_stream_until_notification_message({method})");
+
+        loop {
+            let message = self.read_jsonrpc_message().await?;
+            match message {
+                JSONRPCMessage::Notification(notification) => {
+                    if notification.method == method {
+                        return Ok(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:?}");
+                }
+            }
+        }
+    }
+}

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

@@ -0,0 +1,47 @@
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::Respond;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+
+/// 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;
+
+    let num_calls = responses.len();
+    let seq_responder = SeqResponder {
+        num_calls: AtomicUsize::new(0),
+        responses,
+    };
+
+    Mock::given(method("POST"))
+        .and(path("/v1/chat/completions"))
+        .respond_with(seq_responder)
+        .expect(num_calls as u64)
+        .mount(&server)
+        .await;
+
+    server
+}
+
+struct SeqResponder {
+    num_calls: AtomicUsize,
+    responses: Vec<String>,
+}
+
+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"),
+            None => panic!("no response for {call_num}"),
+        }
+    }
+}

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

@@ -0,0 +1,95 @@
+use serde_json::json;
+use std::path::Path;
+
+pub fn create_shell_sse_response(
+    command: Vec<String>,
+    workdir: Option<&Path>,
+    timeout_ms: Option<u64>,
+    call_id: &str,
+) -> anyhow::Result<String> {
+    // The `arguments`` for the `shell` tool is a serialized JSON object.
+    let tool_call_arguments = serde_json::to_string(&json!({
+        "command": command,
+        "workdir": workdir.map(|w| w.to_string_lossy()),
+        "timeout": timeout_ms
+    }))?;
+    let tool_call = json!({
+        "choices": [
+            {
+                "delta": {
+                    "tool_calls": [
+                        {
+                            "id": call_id,
+                            "function": {
+                                "name": "shell",
+                                "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)
+}
+
+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)
+}
+
+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 shell_command = format!("apply_patch <<'EOF'\n{patch_content}\nEOF");
+    let tool_call_arguments = serde_json::to_string(&json!({
+        "command": ["bash", "-lc", shell_command]
+    }))?;
+
+    let tool_call = json!({
+        "choices": [
+            {
+                "delta": {
+                    "tool_calls": [
+                        {
+                            "id": call_id,
+                            "function": {
+                                "name": "shell",
+                                "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)
+}

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

@@ -1,12 +1,12 @@
 use std::path::Path;
 
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_core::ARCHIVED_SESSIONS_SUBDIR;
 use codex_protocol::mcp_protocol::ArchiveConversationParams;
 use codex_protocol::mcp_protocol::ArchiveConversationResponse;
 use codex_protocol::mcp_protocol::NewConversationParams;
 use codex_protocol::mcp_protocol::NewConversationResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use tempfile::TempDir;

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

@@ -1,12 +1,12 @@
 use std::path::Path;
 
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_protocol::mcp_protocol::AuthMode;
 use codex_protocol::mcp_protocol::GetAuthStatusParams;
 use codex_protocol::mcp_protocol::GetAuthStatusResponse;
 use codex_protocol::mcp_protocol::LoginApiKeyParams;
 use codex_protocol::mcp_protocol::LoginApiKeyResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use pretty_assertions::assert_eq;

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

@@ -1,5 +1,10 @@
 use std::path::Path;
 
+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_shell_sse_response;
+use app_test_support::to_response;
 use codex_core::protocol::AskForApproval;
 use codex_core::protocol::SandboxPolicy;
 use codex_core::protocol_config_types::ReasoningEffort;
@@ -16,11 +21,6 @@ use codex_protocol::mcp_protocol::SendUserMessageParams;
 use codex_protocol::mcp_protocol::SendUserMessageResponse;
 use codex_protocol::mcp_protocol::SendUserTurnParams;
 use codex_protocol::mcp_protocol::SendUserTurnResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::create_final_assistant_message_sse_response;
-use mcp_test_support::create_mock_chat_completions_server;
-use mcp_test_support::create_shell_sse_response;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCNotification;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;

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

@@ -1,6 +1,8 @@
 use std::collections::HashMap;
 use std::path::Path;
 
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_core::protocol::AskForApproval;
 use codex_protocol::config_types::ReasoningEffort;
 use codex_protocol::config_types::ReasoningSummary;
@@ -11,8 +13,6 @@ use codex_protocol::mcp_protocol::Profile;
 use codex_protocol::mcp_protocol::SandboxSettings;
 use codex_protocol::mcp_protocol::Tools;
 use codex_protocol::mcp_protocol::UserSavedConfig;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use pretty_assertions::assert_eq;

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

@@ -1,5 +1,9 @@
 use std::path::Path;
 
+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_protocol::mcp_protocol::AddConversationListenerParams;
 use codex_protocol::mcp_protocol::AddConversationSubscriptionResponse;
 use codex_protocol::mcp_protocol::InputItem;
@@ -7,10 +11,6 @@ use codex_protocol::mcp_protocol::NewConversationParams;
 use codex_protocol::mcp_protocol::NewConversationResponse;
 use codex_protocol::mcp_protocol::SendUserMessageParams;
 use codex_protocol::mcp_protocol::SendUserMessageResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::create_final_assistant_message_sse_response;
-use mcp_test_support::create_mock_chat_completions_server;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use pretty_assertions::assert_eq;

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

@@ -0,0 +1,104 @@
+use app_test_support::McpProcess;
+use mcp_types::JSONRPCResponse;
+use mcp_types::RequestId;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+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 test_fuzzy_file_search_sorts_and_includes_indices() {
+    // Prepare a temporary Codex home and a separate root with test files.
+    let codex_home = TempDir::new().expect("create temp codex home");
+    let root = TempDir::new().expect("create temp search root");
+
+    // Create files designed to have deterministic ordering for query "abc".
+    std::fs::write(root.path().join("abc"), "x").expect("write file abc");
+    std::fs::write(root.path().join("abcde"), "x").expect("write file abcx");
+    std::fs::write(root.path().join("abexy"), "x").expect("write file abcx");
+    std::fs::write(root.path().join("zzz.txt"), "x").expect("write file zzz");
+
+    // Start MCP server and initialize.
+    let mut mcp = McpProcess::new(codex_home.path()).await.expect("spawn mcp");
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize())
+        .await
+        .expect("init timeout")
+        .expect("init failed");
+
+    let root_path = root.path().to_string_lossy().to_string();
+    // Send fuzzyFileSearch request.
+    let request_id = mcp
+        .send_fuzzy_file_search_request("abe", vec![root_path.clone()], None)
+        .await
+        .expect("send fuzzyFileSearch");
+
+    // Read response and verify shape and ordering.
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await
+    .expect("fuzzyFileSearch timeout")
+    .expect("fuzzyFileSearch resp");
+
+    let value = resp.result;
+    assert_eq!(
+        value,
+        json!({
+            "files": [
+                { "root": root_path.clone(), "path": "abexy", "score": 88, "indices": [0, 1, 2] },
+                { "root": root_path.clone(), "path": "abcde", "score": 74, "indices": [0, 1, 4] },
+            ]
+        })
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_fuzzy_file_search_accepts_cancellation_token() {
+    let codex_home = TempDir::new().expect("create temp codex home");
+    let root = TempDir::new().expect("create temp search root");
+
+    std::fs::write(root.path().join("alpha.txt"), "contents").expect("write alpha");
+
+    let mut mcp = McpProcess::new(codex_home.path()).await.expect("spawn mcp");
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize())
+        .await
+        .expect("init timeout")
+        .expect("init failed");
+
+    let root_path = root.path().to_string_lossy().to_string();
+    let request_id = mcp
+        .send_fuzzy_file_search_request("alp", vec![root_path.clone()], None)
+        .await
+        .expect("send fuzzyFileSearch");
+
+    let request_id_2 = mcp
+        .send_fuzzy_file_search_request(
+            "alp",
+            vec![root_path.clone()],
+            Some(request_id.to_string()),
+        )
+        .await
+        .expect("send fuzzyFileSearch");
+
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id_2)),
+    )
+    .await
+    .expect("fuzzyFileSearch timeout")
+    .expect("fuzzyFileSearch resp");
+
+    let files = resp
+        .result
+        .get("files")
+        .and_then(|value| value.as_array())
+        .cloned()
+        .expect("files array");
+
+    assert_eq!(files.len(), 1);
+    assert_eq!(files[0]["root"], root_path);
+    assert_eq!(files[0]["path"], "alpha.txt");
+}

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

@@ -1,5 +1,5 @@
 #![cfg(unix)]
-// Support code lives in the `mcp_test_support` crate under tests/common.
+// Support code lives in the `app_test_support` crate under tests/common.
 
 use std::path::Path;
 
@@ -11,22 +11,22 @@ use codex_protocol::mcp_protocol::NewConversationParams;
 use codex_protocol::mcp_protocol::NewConversationResponse;
 use codex_protocol::mcp_protocol::SendUserMessageParams;
 use codex_protocol::mcp_protocol::SendUserMessageResponse;
-use core_test_support::non_sandbox_test;
+use core_test_support::skip_if_no_network;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use tempfile::TempDir;
 use tokio::time::timeout;
 
-use mcp_test_support::McpProcess;
-use mcp_test_support::create_mock_chat_completions_server;
-use mcp_test_support::create_shell_sse_response;
-use mcp_test_support::to_response;
+use app_test_support::McpProcess;
+use app_test_support::create_mock_chat_completions_server;
+use app_test_support::create_shell_sse_response;
+use app_test_support::to_response;
 
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn test_shell_command_interruption() {
-    non_sandbox_test!();
+    skip_if_no_network!();
 
     if let Err(err) = shell_command_interruption().await {
         panic!("failure: {err}");

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

@@ -1,13 +1,15 @@
 use std::fs;
 use std::path::Path;
 
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_protocol::mcp_protocol::ListConversationsParams;
 use codex_protocol::mcp_protocol::ListConversationsResponse;
 use codex_protocol::mcp_protocol::NewConversationParams; // reused for overrides shape
 use codex_protocol::mcp_protocol::ResumeConversationParams;
 use codex_protocol::mcp_protocol::ResumeConversationResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
+use codex_protocol::mcp_protocol::ServerNotification;
+use codex_protocol::mcp_protocol::SessionConfiguredNotification;
 use mcp_types::JSONRPCNotification;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
@@ -111,23 +113,28 @@ async fn test_list_and_resume_conversations() {
         .await
         .expect("send resumeConversation");
 
-    // Expect a codex/event notification with msg.type == session_configured
+    // Expect a codex/event notification with msg.type == sessionConfigured
     let notification: JSONRPCNotification = timeout(
         DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("codex/event"),
+        mcp.read_stream_until_notification_message("sessionConfigured"),
     )
     .await
-    .expect("session_configured notification timeout")
-    .expect("session_configured notification");
-    // Basic shape assertion: ensure event type is session_configured
-    let msg_type = notification
-        .params
-        .as_ref()
-        .and_then(|p| p.get("msg"))
-        .and_then(|m| m.get("type"))
-        .and_then(|t| t.as_str())
-        .unwrap_or("");
-    assert_eq!(msg_type, "session_configured");
+    .expect("sessionConfigured notification timeout")
+    .expect("sessionConfigured notification");
+    let session_configured: ServerNotification = notification
+        .try_into()
+        .expect("deserialize sessionConfigured notification");
+    // Basic shape assertion: ensure event type is sessionConfigured
+    let ServerNotification::SessionConfigured(SessionConfiguredNotification {
+        model,
+        rollout_path,
+        ..
+    }) = session_configured
+    else {
+        unreachable!("expected sessionConfigured notification");
+    };
+    assert_eq!(model, "o3");
+    assert_eq!(items[0].path.clone(), rollout_path);
 
     // Then the response for resumeConversation
     let resume_resp: JSONRPCResponse = timeout(

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

@@ -1,6 +1,8 @@
 use std::path::Path;
 use std::time::Duration;
 
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_login::login_with_api_key;
 use codex_protocol::mcp_protocol::CancelLoginChatGptParams;
 use codex_protocol::mcp_protocol::CancelLoginChatGptResponse;
@@ -8,8 +10,6 @@ use codex_protocol::mcp_protocol::GetAuthStatusParams;
 use codex_protocol::mcp_protocol::GetAuthStatusResponse;
 use codex_protocol::mcp_protocol::LoginChatGptResponse;
 use codex_protocol::mcp_protocol::LogoutChatGptResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use tempfile::TempDir;

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

@@ -0,0 +1,13 @@
+mod archive_conversation;
+mod auth;
+mod codex_message_processor_flow;
+mod config;
+mod create_conversation;
+mod fuzzy_file_search;
+mod interrupt;
+mod list_resume;
+mod login;
+mod send_message;
+mod set_default_model;
+mod user_agent;
+mod user_info;

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

@@ -1,5 +1,9 @@
 use std::path::Path;
 
+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_protocol::mcp_protocol::AddConversationListenerParams;
 use codex_protocol::mcp_protocol::AddConversationSubscriptionResponse;
 use codex_protocol::mcp_protocol::ConversationId;
@@ -8,10 +12,6 @@ use codex_protocol::mcp_protocol::NewConversationParams;
 use codex_protocol::mcp_protocol::NewConversationResponse;
 use codex_protocol::mcp_protocol::SendUserMessageParams;
 use codex_protocol::mcp_protocol::SendUserMessageResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::create_final_assistant_message_sse_response;
-use mcp_test_support::create_mock_chat_completions_server;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCNotification;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;

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

@@ -1,10 +1,10 @@
 use std::path::Path;
 
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_core::config::ConfigToml;
 use codex_protocol::mcp_protocol::SetDefaultModelParams;
 use codex_protocol::mcp_protocol::SetDefaultModelResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use pretty_assertions::assert_eq;

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

@@ -1,6 +1,6 @@
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use codex_protocol::mcp_protocol::GetUserAgentResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use pretty_assertions::assert_eq;
@@ -35,7 +35,7 @@ async fn get_user_agent_returns_current_codex_user_agent() {
 
     let os_info = os_info::get();
     let user_agent = format!(
-        "codex_cli_rs/0.0.0 ({} {}; {}) {} (elicitation test; 0.0.0)",
+        "codex_cli_rs/0.0.0 ({} {}; {}) {} (codex-app-server-tests; 0.1.0)",
         os_info.os_type(),
         os_info.version(),
         os_info.architecture().unwrap_or("unknown"),

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

@@ -1,6 +1,8 @@
 use std::time::Duration;
 
 use anyhow::Context;
+use app_test_support::McpProcess;
+use app_test_support::to_response;
 use base64::Engine;
 use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use codex_core::auth::AuthDotJson;
@@ -9,8 +11,6 @@ use codex_core::auth::write_auth_json;
 use codex_core::token_data::IdTokenInfo;
 use codex_core::token_data::TokenData;
 use codex_protocol::mcp_protocol::UserInfoResponse;
-use mcp_test_support::McpProcess;
-use mcp_test_support::to_response;
 use mcp_types::JSONRPCResponse;
 use mcp_types::RequestId;
 use pretty_assertions::assert_eq;

codex-rs/apply-patch/Cargo.toml

@@ -20,7 +20,6 @@ similar = { workspace = true }
 thiserror = { workspace = true }
 tree-sitter = { workspace = true }
 tree-sitter-bash = { workspace = true }
-once_cell = { workspace = true }
 
 [dev-dependencies]
 assert_cmd = { workspace = true }

codex-rs/apply-patch/src/lib.rs

@@ -6,10 +6,10 @@ use std::collections::HashMap;
 use std::path::Path;
 use std::path::PathBuf;
 use std::str::Utf8Error;
+use std::sync::LazyLock;
 
 use anyhow::Context;
 use anyhow::Result;
-use once_cell::sync::Lazy;
 pub use parser::Hunk;
 pub use parser::ParseError;
 use parser::ParseError::*;
@@ -351,7 +351,7 @@ fn extract_apply_patch_from_bash(
     // also run an arbitrary query against the AST. This is useful for understanding
     // how tree-sitter parses the script and whether the query syntax is correct. Be sure
     // to test both positive and negative cases.
-    static APPLY_PATCH_QUERY: Lazy<Query> = Lazy::new(|| {
+    static APPLY_PATCH_QUERY: LazyLock<Query> = LazyLock::new(|| {
         let language = BASH.into();
         #[expect(clippy::expect_used)]
         Query::new(

codex-rs/backend-client/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "codex-backend-client"
+version = "0.0.0"
+edition = "2024"
+publish = false
+
+[lib]
+path = "src/lib.rs"
+
+[dependencies]
+anyhow = "1"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+codex-backend-openapi-models = { path = "../codex-backend-openapi-models" }
+
+[dev-dependencies]
+pretty_assertions = "1"

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

@@ -0,0 +1,244 @@
+use crate::types::CodeTaskDetailsResponse;
+use crate::types::PaginatedListTaskListItem;
+use crate::types::TurnAttemptsSiblingTurnsResponse;
+use anyhow::Result;
+use reqwest::header::AUTHORIZATION;
+use reqwest::header::CONTENT_TYPE;
+use reqwest::header::HeaderMap;
+use reqwest::header::HeaderName;
+use reqwest::header::HeaderValue;
+use reqwest::header::USER_AGENT;
+use serde::de::DeserializeOwned;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PathStyle {
+    /// /api/codex/…
+    CodexApi,
+    /// /wham/…
+    ChatGptApi,
+}
+
+impl PathStyle {
+    pub fn from_base_url(base_url: &str) -> Self {
+        if base_url.contains("/backend-api") {
+            PathStyle::ChatGptApi
+        } else {
+            PathStyle::CodexApi
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct Client {
+    base_url: String,
+    http: reqwest::Client,
+    bearer_token: Option<String>,
+    user_agent: Option<HeaderValue>,
+    chatgpt_account_id: Option<String>,
+    path_style: PathStyle,
+}
+
+impl Client {
+    pub fn new(base_url: impl Into<String>) -> Result<Self> {
+        let mut base_url = base_url.into();
+        // Normalize common ChatGPT hostnames to include /backend-api so we hit the WHAM paths.
+        // Also trim trailing slashes for consistent URL building.
+        while base_url.ends_with('/') {
+            base_url.pop();
+        }
+        if (base_url.starts_with("https://chatgpt.com")
+            || base_url.starts_with("https://chat.openai.com"))
+            && !base_url.contains("/backend-api")
+        {
+            base_url = format!("{base_url}/backend-api");
+        }
+        let http = reqwest::Client::builder().build()?;
+        let path_style = PathStyle::from_base_url(&base_url);
+        Ok(Self {
+            base_url,
+            http,
+            bearer_token: None,
+            user_agent: None,
+            chatgpt_account_id: None,
+            path_style,
+        })
+    }
+
+    pub fn with_bearer_token(mut self, token: impl Into<String>) -> Self {
+        self.bearer_token = Some(token.into());
+        self
+    }
+
+    pub fn with_user_agent(mut self, ua: impl Into<String>) -> Self {
+        if let Ok(hv) = HeaderValue::from_str(&ua.into()) {
+            self.user_agent = Some(hv);
+        }
+        self
+    }
+
+    pub fn with_chatgpt_account_id(mut self, account_id: impl Into<String>) -> Self {
+        self.chatgpt_account_id = Some(account_id.into());
+        self
+    }
+
+    pub fn with_path_style(mut self, style: PathStyle) -> Self {
+        self.path_style = style;
+        self
+    }
+
+    fn headers(&self) -> HeaderMap {
+        let mut h = HeaderMap::new();
+        if let Some(ua) = &self.user_agent {
+            h.insert(USER_AGENT, ua.clone());
+        } else {
+            h.insert(USER_AGENT, HeaderValue::from_static("codex-cli"));
+        }
+        if let Some(token) = &self.bearer_token {
+            let value = format!("Bearer {token}");
+            if let Ok(hv) = HeaderValue::from_str(&value) {
+                h.insert(AUTHORIZATION, hv);
+            }
+        }
+        if let Some(acc) = &self.chatgpt_account_id
+            && let Ok(name) = HeaderName::from_bytes(b"ChatGPT-Account-Id")
+            && let Ok(hv) = HeaderValue::from_str(acc)
+        {
+            h.insert(name, hv);
+        }
+        h
+    }
+
+    async fn exec_request(
+        &self,
+        req: reqwest::RequestBuilder,
+        method: &str,
+        url: &str,
+    ) -> Result<(String, String)> {
+        let res = req.send().await?;
+        let status = res.status();
+        let ct = res
+            .headers()
+            .get(CONTENT_TYPE)
+            .and_then(|v| v.to_str().ok())
+            .unwrap_or("")
+            .to_string();
+        let body = res.text().await.unwrap_or_default();
+        if !status.is_success() {
+            anyhow::bail!("{method} {url} failed: {status}; content-type={ct}; body={body}");
+        }
+        Ok((body, ct))
+    }
+
+    fn decode_json<T: DeserializeOwned>(&self, url: &str, ct: &str, body: &str) -> Result<T> {
+        match serde_json::from_str::<T>(body) {
+            Ok(v) => Ok(v),
+            Err(e) => {
+                anyhow::bail!("Decode error for {url}: {e}; content-type={ct}; body={body}");
+            }
+        }
+    }
+
+    pub async fn list_tasks(
+        &self,
+        limit: Option<i32>,
+        task_filter: Option<&str>,
+        environment_id: Option<&str>,
+    ) -> Result<PaginatedListTaskListItem> {
+        let url = match self.path_style {
+            PathStyle::CodexApi => format!("{}/api/codex/tasks/list", self.base_url),
+            PathStyle::ChatGptApi => format!("{}/wham/tasks/list", self.base_url),
+        };
+        let req = self.http.get(&url).headers(self.headers());
+        let req = if let Some(lim) = limit {
+            req.query(&[("limit", lim)])
+        } else {
+            req
+        };
+        let req = if let Some(tf) = task_filter {
+            req.query(&[("task_filter", tf)])
+        } else {
+            req
+        };
+        let req = if let Some(id) = environment_id {
+            req.query(&[("environment_id", id)])
+        } else {
+            req
+        };
+        let (body, ct) = self.exec_request(req, "GET", &url).await?;
+        self.decode_json::<PaginatedListTaskListItem>(&url, &ct, &body)
+    }
+
+    pub async fn get_task_details(&self, task_id: &str) -> Result<CodeTaskDetailsResponse> {
+        let (parsed, _body, _ct) = self.get_task_details_with_body(task_id).await?;
+        Ok(parsed)
+    }
+
+    pub async fn get_task_details_with_body(
+        &self,
+        task_id: &str,
+    ) -> Result<(CodeTaskDetailsResponse, String, String)> {
+        let url = match self.path_style {
+            PathStyle::CodexApi => format!("{}/api/codex/tasks/{}", self.base_url, task_id),
+            PathStyle::ChatGptApi => format!("{}/wham/tasks/{}", self.base_url, task_id),
+        };
+        let req = self.http.get(&url).headers(self.headers());
+        let (body, ct) = self.exec_request(req, "GET", &url).await?;
+        let parsed: CodeTaskDetailsResponse = self.decode_json(&url, &ct, &body)?;
+        Ok((parsed, body, ct))
+    }
+
+    pub async fn list_sibling_turns(
+        &self,
+        task_id: &str,
+        turn_id: &str,
+    ) -> Result<TurnAttemptsSiblingTurnsResponse> {
+        let url = match self.path_style {
+            PathStyle::CodexApi => format!(
+                "{}/api/codex/tasks/{}/turns/{}/sibling_turns",
+                self.base_url, task_id, turn_id
+            ),
+            PathStyle::ChatGptApi => format!(
+                "{}/wham/tasks/{}/turns/{}/sibling_turns",
+                self.base_url, task_id, turn_id
+            ),
+        };
+        let req = self.http.get(&url).headers(self.headers());
+        let (body, ct) = self.exec_request(req, "GET", &url).await?;
+        self.decode_json::<TurnAttemptsSiblingTurnsResponse>(&url, &ct, &body)
+    }
+
+    /// Create a new task (user turn) by POSTing to the appropriate backend path
+    /// based on `path_style`. Returns the created task id.
+    pub async fn create_task(&self, request_body: serde_json::Value) -> Result<String> {
+        let url = match self.path_style {
+            PathStyle::CodexApi => format!("{}/api/codex/tasks", self.base_url),
+            PathStyle::ChatGptApi => format!("{}/wham/tasks", self.base_url),
+        };
+        let req = self
+            .http
+            .post(&url)
+            .headers(self.headers())
+            .header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
+            .json(&request_body);
+        let (body, ct) = self.exec_request(req, "POST", &url).await?;
+        // Extract id from JSON: prefer `task.id`; fallback to top-level `id` when present.
+        match serde_json::from_str::<serde_json::Value>(&body) {
+            Ok(v) => {
+                if let Some(id) = v
+                    .get("task")
+                    .and_then(|t| t.get("id"))
+                    .and_then(|s| s.as_str())
+                {
+                    Ok(id.to_string())
+                } else if let Some(id) = v.get("id").and_then(|s| s.as_str()) {
+                    Ok(id.to_string())
+                } else {
+                    anyhow::bail!(
+                        "POST {url} succeeded but no task id found; content-type={ct}; body={body}"
+                    );
+                }
+            }
+            Err(e) => anyhow::bail!("Decode error for {url}: {e}; content-type={ct}; body={body}"),
+        }
+    }
+}

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

@@ -0,0 +1,9 @@
+mod client;
+pub mod types;
+
+pub use client::Client;
+pub use types::CodeTaskDetailsResponse;
+pub use types::CodeTaskDetailsResponseExt;
+pub use types::PaginatedListTaskListItem;
+pub use types::TaskListItem;
+pub use types::TurnAttemptsSiblingTurnsResponse;

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

@@ -0,0 +1,369 @@
+pub use codex_backend_openapi_models::models::PaginatedListTaskListItem;
+pub use codex_backend_openapi_models::models::TaskListItem;
+
+use serde::Deserialize;
+use serde::de::Deserializer;
+use serde_json::Value;
+use std::collections::HashMap;
+
+/// Hand-rolled models for the Cloud Tasks task-details response.
+/// The generated OpenAPI models are pretty bad. This is a half-step
+/// towards hand-rolling them.
+#[derive(Clone, Debug, Deserialize)]
+pub struct CodeTaskDetailsResponse {
+    #[serde(default)]
+    pub current_user_turn: Option<Turn>,
+    #[serde(default)]
+    pub current_assistant_turn: Option<Turn>,
+    #[serde(default)]
+    pub current_diff_task_turn: Option<Turn>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct Turn {
+    #[serde(default)]
+    pub id: Option<String>,
+    #[serde(default)]
+    pub attempt_placement: Option<i64>,
+    #[serde(default, rename = "turn_status")]
+    pub turn_status: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_vec")]
+    pub sibling_turn_ids: Vec<String>,
+    #[serde(default, deserialize_with = "deserialize_vec")]
+    pub input_items: Vec<TurnItem>,
+    #[serde(default, deserialize_with = "deserialize_vec")]
+    pub output_items: Vec<TurnItem>,
+    #[serde(default)]
+    pub worklog: Option<Worklog>,
+    #[serde(default)]
+    pub error: Option<TurnError>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct TurnItem {
+    #[serde(rename = "type", default)]
+    pub kind: String,
+    #[serde(default)]
+    pub role: Option<String>,
+    #[serde(default, deserialize_with = "deserialize_vec")]
+    pub content: Vec<ContentFragment>,
+    #[serde(default)]
+    pub diff: Option<String>,
+    #[serde(default)]
+    pub output_diff: Option<DiffPayload>,
+}
+
+#[derive(Clone, Debug, Deserialize)]
+#[serde(untagged)]
+pub enum ContentFragment {
+    Structured(StructuredContent),
+    Text(String),
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct StructuredContent {
+    #[serde(rename = "content_type", default)]
+    pub content_type: Option<String>,
+    #[serde(default)]
+    pub text: Option<String>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct DiffPayload {
+    #[serde(default)]
+    pub diff: Option<String>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct Worklog {
+    #[serde(default, deserialize_with = "deserialize_vec")]
+    pub messages: Vec<WorklogMessage>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct WorklogMessage {
+    #[serde(default)]
+    pub author: Option<Author>,
+    #[serde(default)]
+    pub content: Option<WorklogContent>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct Author {
+    #[serde(default)]
+    pub role: Option<String>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct WorklogContent {
+    #[serde(default)]
+    pub parts: Vec<ContentFragment>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct TurnError {
+    #[serde(default)]
+    pub code: Option<String>,
+    #[serde(default)]
+    pub message: Option<String>,
+}
+
+impl ContentFragment {
+    fn text(&self) -> Option<&str> {
+        match self {
+            ContentFragment::Structured(inner) => {
+                if inner
+                    .content_type
+                    .as_deref()
+                    .map(|ct| ct.eq_ignore_ascii_case("text"))
+                    .unwrap_or(false)
+                {
+                    inner.text.as_deref().filter(|s| !s.is_empty())
+                } else {
+                    None
+                }
+            }
+            ContentFragment::Text(raw) => {
+                if raw.trim().is_empty() {
+                    None
+                } else {
+                    Some(raw.as_str())
+                }
+            }
+        }
+    }
+}
+
+impl TurnItem {
+    fn text_values(&self) -> Vec<String> {
+        self.content
+            .iter()
+            .filter_map(|fragment| fragment.text().map(str::to_string))
+            .collect()
+    }
+
+    fn diff_text(&self) -> Option<String> {
+        if self.kind == "output_diff" {
+            if let Some(diff) = &self.diff
+                && !diff.is_empty()
+            {
+                return Some(diff.clone());
+            }
+        } else if self.kind == "pr"
+            && let Some(payload) = &self.output_diff
+            && let Some(diff) = &payload.diff
+            && !diff.is_empty()
+        {
+            return Some(diff.clone());
+        }
+        None
+    }
+}
+
+impl Turn {
+    fn unified_diff(&self) -> Option<String> {
+        self.output_items.iter().find_map(TurnItem::diff_text)
+    }
+
+    fn message_texts(&self) -> Vec<String> {
+        let mut out: Vec<String> = self
+            .output_items
+            .iter()
+            .filter(|item| item.kind == "message")
+            .flat_map(TurnItem::text_values)
+            .collect();
+
+        if let Some(log) = &self.worklog {
+            for message in &log.messages {
+                if message.is_assistant() {
+                    out.extend(message.text_values());
+                }
+            }
+        }
+
+        out
+    }
+
+    fn user_prompt(&self) -> Option<String> {
+        let parts: Vec<String> = self
+            .input_items
+            .iter()
+            .filter(|item| item.kind == "message")
+            .filter(|item| {
+                item.role
+                    .as_deref()
+                    .map(|r| r.eq_ignore_ascii_case("user"))
+                    .unwrap_or(true)
+            })
+            .flat_map(TurnItem::text_values)
+            .collect();
+
+        if parts.is_empty() {
+            None
+        } else {
+            Some(parts.join(
+                "
+
+",
+            ))
+        }
+    }
+
+    fn error_summary(&self) -> Option<String> {
+        self.error.as_ref().and_then(TurnError::summary)
+    }
+}
+
+impl WorklogMessage {
+    fn is_assistant(&self) -> bool {
+        self.author
+            .as_ref()
+            .and_then(|a| a.role.as_deref())
+            .map(|role| role.eq_ignore_ascii_case("assistant"))
+            .unwrap_or(false)
+    }
+
+    fn text_values(&self) -> Vec<String> {
+        self.content
+            .as_ref()
+            .map(|content| {
+                content
+                    .parts
+                    .iter()
+                    .filter_map(|fragment| fragment.text().map(str::to_string))
+                    .collect()
+            })
+            .unwrap_or_default()
+    }
+}
+
+impl TurnError {
+    fn summary(&self) -> Option<String> {
+        let code = self.code.as_deref().unwrap_or("");
+        let message = self.message.as_deref().unwrap_or("");
+        match (code.is_empty(), message.is_empty()) {
+            (true, true) => None,
+            (false, true) => Some(code.to_string()),
+            (true, false) => Some(message.to_string()),
+            (false, false) => Some(format!("{code}: {message}")),
+        }
+    }
+}
+
+pub trait CodeTaskDetailsResponseExt {
+    /// Attempt to extract a unified diff string from the assistant or diff turn.
+    fn unified_diff(&self) -> Option<String>;
+    /// Extract assistant text output messages (no diff) from current turns.
+    fn assistant_text_messages(&self) -> Vec<String>;
+    /// Extract the user's prompt text from the current user turn, when present.
+    fn user_text_prompt(&self) -> Option<String>;
+    /// Extract an assistant error message (if the turn failed and provided one).
+    fn assistant_error_message(&self) -> Option<String>;
+}
+
+impl CodeTaskDetailsResponseExt for CodeTaskDetailsResponse {
+    fn unified_diff(&self) -> Option<String> {
+        [
+            self.current_diff_task_turn.as_ref(),
+            self.current_assistant_turn.as_ref(),
+        ]
+        .into_iter()
+        .flatten()
+        .find_map(Turn::unified_diff)
+    }
+
+    fn assistant_text_messages(&self) -> Vec<String> {
+        let mut out = Vec::new();
+        for turn in [
+            self.current_diff_task_turn.as_ref(),
+            self.current_assistant_turn.as_ref(),
+        ]
+        .into_iter()
+        .flatten()
+        {
+            out.extend(turn.message_texts());
+        }
+        out
+    }
+
+    fn user_text_prompt(&self) -> Option<String> {
+        self.current_user_turn.as_ref().and_then(Turn::user_prompt)
+    }
+
+    fn assistant_error_message(&self) -> Option<String> {
+        self.current_assistant_turn
+            .as_ref()
+            .and_then(Turn::error_summary)
+    }
+}
+
+fn deserialize_vec<'de, D, T>(deserializer: D) -> Result<Vec<T>, D::Error>
+where
+    D: Deserializer<'de>,
+    T: Deserialize<'de>,
+{
+    Option::<Vec<T>>::deserialize(deserializer).map(|opt| opt.unwrap_or_default())
+}
+
+#[derive(Clone, Debug, Deserialize)]
+pub struct TurnAttemptsSiblingTurnsResponse {
+    #[serde(default)]
+    pub sibling_turns: Vec<HashMap<String, Value>>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+
+    fn fixture(name: &str) -> CodeTaskDetailsResponse {
+        let json = match name {
+            "diff" => include_str!("../tests/fixtures/task_details_with_diff.json"),
+            "error" => include_str!("../tests/fixtures/task_details_with_error.json"),
+            other => panic!("unknown fixture {other}"),
+        };
+        serde_json::from_str(json).expect("fixture should deserialize")
+    }
+
+    #[test]
+    fn unified_diff_prefers_current_diff_task_turn() {
+        let details = fixture("diff");
+        let diff = details.unified_diff().expect("diff present");
+        assert!(diff.contains("diff --git"));
+    }
+
+    #[test]
+    fn unified_diff_falls_back_to_pr_output_diff() {
+        let details = fixture("error");
+        let diff = details.unified_diff().expect("diff from pr output");
+        assert!(diff.contains("lib.rs"));
+    }
+
+    #[test]
+    fn assistant_text_messages_extracts_text_content() {
+        let details = fixture("diff");
+        let messages = details.assistant_text_messages();
+        assert_eq!(messages, vec!["Assistant response".to_string()]);
+    }
+
+    #[test]
+    fn user_text_prompt_joins_parts_with_spacing() {
+        let details = fixture("diff");
+        let prompt = details.user_text_prompt().expect("prompt present");
+        assert_eq!(
+            prompt,
+            "First line
+
+Second line"
+        );
+    }
+
+    #[test]
+    fn assistant_error_message_combines_code_and_message() {
+        let details = fixture("error");
+        let msg = details
+            .assistant_error_message()
+            .expect("error should be present");
+        assert_eq!(msg, "APPLY_FAILED: Patch could not be applied");
+    }
+}

codex-rs/backend-client/tests/fixtures/task_details_with_diff.json

@@ -0,0 +1,38 @@
+{
+  "task": {
+    "id": "task_123",
+    "title": "Refactor cloud task client",
+    "archived": false,
+    "external_pull_requests": []
+  },
+  "current_user_turn": {
+    "input_items": [
+      {
+        "type": "message",
+        "role": "user",
+        "content": [
+          { "content_type": "text", "text": "First line" },
+          { "content_type": "text", "text": "Second line" }
+        ]
+      }
+    ]
+  },
+  "current_assistant_turn": {
+    "output_items": [
+      {
+        "type": "message",
+        "content": [
+          { "content_type": "text", "text": "Assistant response" }
+        ]
+      }
+    ]
+  },
+  "current_diff_task_turn": {
+    "output_items": [
+      {
+        "type": "output_diff",
+        "diff": "diff --git a/src/main.rs b/src/main.rs\n+fn main() { println!(\"hi\"); }\n"
+      }
+    ]
+  }
+}

codex-rs/backend-client/tests/fixtures/task_details_with_error.json

@@ -0,0 +1,22 @@
+{
+  "task": {
+    "id": "task_456",
+    "title": "Investigate failure",
+    "archived": false,
+    "external_pull_requests": []
+  },
+  "current_assistant_turn": {
+    "output_items": [
+      {
+        "type": "pr",
+        "output_diff": {
+          "diff": "diff --git a/lib.rs b/lib.rs\n+pub fn hello() {}\n"
+        }
+      }
+    ],
+    "error": {
+      "code": "APPLY_FAILED",
+      "message": "Patch could not be applied"
+    }
+  }
+}

codex-rs/chatgpt/Cargo.toml

@@ -14,6 +14,7 @@ codex-core = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
+codex-git-apply = { path = "../git-apply" }
 
 [dev-dependencies]
 tempfile = { workspace = true }

codex-rs/chatgpt/src/apply_command.rs

@@ -56,46 +56,24 @@ pub async fn apply_diff_from_task(
 }
 
 async fn apply_diff(diff: &str, cwd: Option<PathBuf>) -> anyhow::Result<()> {
-    let mut cmd = tokio::process::Command::new("git");
-    if let Some(cwd) = cwd {
-        cmd.current_dir(cwd);
-    }
-    let toplevel_output = cmd
-        .args(vec!["rev-parse", "--show-toplevel"])
-        .output()
-        .await?;
-
-    if !toplevel_output.status.success() {
-        anyhow::bail!("apply must be run from a git repository.");
-    }
-
-    let repo_root = String::from_utf8(toplevel_output.stdout)?
-        .trim()
-        .to_string();
-
-    let mut git_apply_cmd = tokio::process::Command::new("git")
-        .args(vec!["apply", "--3way"])
-        .current_dir(&repo_root)
-        .stdin(std::process::Stdio::piped())
-        .stdout(std::process::Stdio::piped())
-        .stderr(std::process::Stdio::piped())
-        .spawn()?;
-
-    if let Some(mut stdin) = git_apply_cmd.stdin.take() {
-        tokio::io::AsyncWriteExt::write_all(&mut stdin, diff.as_bytes()).await?;
-        drop(stdin);
-    }
-
-    let output = git_apply_cmd.wait_with_output().await?;
-
-    if !output.status.success() {
+    let cwd = cwd.unwrap_or(std::env::current_dir().unwrap_or_else(|_| std::env::temp_dir()));
+    let req = codex_git_apply::ApplyGitRequest {
+        cwd,
+        diff: diff.to_string(),
+        revert: false,
+        preflight: false,
+    };
+    let res = codex_git_apply::apply_git_patch(&req)?;
+    if res.exit_code != 0 {
         anyhow::bail!(
-            "Git apply failed with status {}: {}",
-            output.status,
-            String::from_utf8_lossy(&output.stderr)
+            "Git apply failed (applied={}, skipped={}, conflicts={})\nstdout:\n{}\nstderr:\n{}",
+            res.applied_paths.len(),
+            res.skipped_paths.len(),
+            res.conflicted_paths.len(),
+            res.stdout,
+            res.stderr
         );
     }
-
     println!("Successfully applied diff");
     Ok(())
 }

codex-rs/chatgpt/src/chatgpt_client.rs

@@ -44,6 +44,6 @@ pub(crate) async fn chatgpt_get_request<T: DeserializeOwned>(
     } else {
         let status = response.status();
         let body = response.text().await.unwrap_or_default();
-        anyhow::bail!("Request failed with status {}: {}", status, body)
+        anyhow::bail!("Request failed with status {status}: {body}")
     }
 }

codex-rs/cli/Cargo.toml

@@ -18,6 +18,7 @@ workspace = true
 anyhow = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 clap_complete = { workspace = true }
+codex-app-server = { workspace = true }
 codex-arg0 = { workspace = true }
 codex-chatgpt = { workspace = true }
 codex-common = { workspace = true, features = ["cli"] }
@@ -25,9 +26,13 @@ codex-core = { workspace = true }
 codex-exec = { workspace = true }
 codex-login = { workspace = true }
 codex-mcp-server = { workspace = true }
+codex-process-hardening = { workspace = true }
 codex-protocol = { workspace = true }
 codex-protocol-ts = { workspace = true }
+codex-responses-api-proxy = { workspace = true }
 codex-tui = { workspace = true }
+codex-cloud-tasks = { path = "../cloud-tasks" }
+ctor = { workspace = true }
 owo-colors = { workspace = true }
 serde_json = { workspace = true }
 supports-color = { workspace = true }

codex-rs/cli/src/login.rs

@@ -6,6 +6,7 @@ use codex_core::auth::logout;
 use codex_core::config::Config;
 use codex_core::config::ConfigOverrides;
 use codex_login::ServerOptions;
+use codex_login::run_device_code_login;
 use codex_login::run_login_server;
 use codex_protocol::mcp_protocol::AuthMode;
 use std::path::PathBuf;
@@ -55,6 +56,32 @@ pub async fn run_login_with_api_key(
     }
 }
 
+/// Login using the OAuth device code flow.
+pub async fn run_login_with_device_code(
+    cli_config_overrides: CliConfigOverrides,
+    issuer_base_url: Option<String>,
+    client_id: Option<String>,
+) -> ! {
+    let config = load_config_or_exit(cli_config_overrides);
+    let mut opts = ServerOptions::new(
+        config.codex_home,
+        client_id.unwrap_or(CLIENT_ID.to_string()),
+    );
+    if let Some(iss) = issuer_base_url {
+        opts.issuer = iss;
+    }
+    match run_device_code_login(opts).await {
+        Ok(()) => {
+            eprintln!("Successfully logged in");
+            std::process::exit(0);
+        }
+        Err(e) => {
+            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);
 

codex-rs/cli/src/main.rs

@@ -10,10 +10,13 @@ use codex_cli::SeatbeltCommand;
 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_logout;
 use codex_cli::proto;
+use codex_cloud_tasks::Cli as CloudTasksCli;
 use codex_common::CliConfigOverrides;
 use codex_exec::Cli as ExecCli;
+use codex_responses_api_proxy::Args as ResponsesApiProxyArgs;
 use codex_tui::AppExitInfo;
 use codex_tui::Cli as TuiCli;
 use owo_colors::OwoColorize;
@@ -65,6 +68,12 @@ enum Subcommand {
     /// [experimental] Run Codex as an MCP server and manage MCP servers.
     Mcp(McpCli),
 
+    /// [experimental] Run the Codex MCP server (stdio transport).
+    McpServer,
+
+    /// [experimental] Run the app server.
+    AppServer,
+
     /// Run the Protocol stream via stdin/stdout
     #[clap(visible_alias = "p")]
     Proto(ProtoCli),
@@ -85,6 +94,13 @@ enum Subcommand {
     /// Internal: generate TypeScript protocol bindings.
     #[clap(hide = true)]
     GenerateTs(GenerateTsCommand),
+    /// [EXPERIMENTAL] Browse tasks from Codex Cloud and apply changes locally.
+    #[clap(name = "cloud", alias = "cloud-tasks")]
+    Cloud(CloudTasksCli),
+
+    /// Internal: run the responses API proxy.
+    #[clap(hide = true)]
+    ResponsesApiProxy(ResponsesApiProxyArgs),
 }
 
 #[derive(Debug, Parser)]
@@ -132,6 +148,20 @@ struct LoginCommand {
     #[arg(long = "api-key", value_name = "API_KEY")]
     api_key: Option<String>,
 
+    /// EXPERIMENTAL: Use device code flow (not yet supported)
+    /// This feature is experimental and may changed in future releases.
+    #[arg(long = "experimental_use-device-code", hide = true)]
+    use_device_code: bool,
+
+    /// EXPERIMENTAL: Use custom OAuth issuer base URL (advanced)
+    /// Override the OAuth issuer base URL (advanced)
+    #[arg(long = "experimental_issuer", value_name = "URL", hide = true)]
+    issuer_base_url: Option<String>,
+
+    /// EXPERIMENTAL: Use custom OAuth client ID (advanced)
+    #[arg(long = "experimental_client-id", value_name = "CLIENT_ID", hide = true)]
+    client_id: Option<String>,
+
     #[command(subcommand)]
     action: Option<LoginSubcommand>,
 }
@@ -181,7 +211,7 @@ fn format_exit_messages(exit_info: AppExitInfo, color_enabled: bool) -> Vec<Stri
         } else {
             resume_cmd
         };
-        lines.push(format!("To continue this session, run {command}."));
+        lines.push(format!("To continue this session, run {command}"));
     }
 
     lines
@@ -194,6 +224,27 @@ fn print_exit_messages(exit_info: AppExitInfo) {
     }
 }
 
+pub(crate) const CODEX_SECURE_MODE_ENV_VAR: &str = "CODEX_SECURE_MODE";
+
+/// As early as possible in the process lifecycle, apply hardening measures
+/// if the CODEX_SECURE_MODE environment variable is set to "1".
+#[ctor::ctor]
+fn pre_main_hardening() {
+    let secure_mode = match std::env::var(CODEX_SECURE_MODE_ENV_VAR) {
+        Ok(value) => value,
+        Err(_) => return,
+    };
+
+    if secure_mode == "1" {
+        codex_process_hardening::pre_main_hardening();
+    }
+
+    // Always clear this env var so child processes don't inherit it.
+    unsafe {
+        std::env::remove_var(CODEX_SECURE_MODE_ENV_VAR);
+    }
+}
+
 fn main() -> anyhow::Result<()> {
     arg0_dispatch_or_else(|codex_linux_sandbox_exe| async move {
         cli_main(codex_linux_sandbox_exe).await?;
@@ -224,10 +275,16 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
             );
             codex_exec::run_main(exec_cli, codex_linux_sandbox_exe).await?;
         }
+        Some(Subcommand::McpServer) => {
+            codex_mcp_server::run_main(codex_linux_sandbox_exe, root_config_overrides).await?;
+        }
         Some(Subcommand::Mcp(mut mcp_cli)) => {
             // Propagate any root-level config overrides (e.g. `-c key=value`).
             prepend_config_flags(&mut mcp_cli.config_overrides, root_config_overrides.clone());
-            mcp_cli.run(codex_linux_sandbox_exe).await?;
+            mcp_cli.run().await?;
+        }
+        Some(Subcommand::AppServer) => {
+            codex_app_server::run_main(codex_linux_sandbox_exe, root_config_overrides).await?;
         }
         Some(Subcommand::Resume(ResumeCommand {
             session_id,
@@ -253,7 +310,14 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
                     run_login_status(login_cli.config_overrides).await;
                 }
                 None => {
-                    if let Some(api_key) = login_cli.api_key {
+                    if login_cli.use_device_code {
+                        run_login_with_device_code(
+                            login_cli.config_overrides,
+                            login_cli.issuer_base_url,
+                            login_cli.client_id,
+                        )
+                        .await;
+                    } else if let Some(api_key) = login_cli.api_key {
                         run_login_with_api_key(login_cli.config_overrides, api_key).await;
                     } else {
                         run_login_with_chatgpt(login_cli.config_overrides).await;
@@ -278,6 +342,13 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
         Some(Subcommand::Completion(completion_cli)) => {
             print_completion(completion_cli);
         }
+        Some(Subcommand::Cloud(mut cloud_cli)) => {
+            prepend_config_flags(
+                &mut cloud_cli.config_overrides,
+                root_config_overrides.clone(),
+            );
+            codex_cloud_tasks::run_main(cloud_cli, codex_linux_sandbox_exe).await?;
+        }
         Some(Subcommand::Debug(debug_args)) => match debug_args.cmd {
             DebugCommand::Seatbelt(mut seatbelt_cli) => {
                 prepend_config_flags(
@@ -309,6 +380,10 @@ async fn cli_main(codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()
             );
             run_apply_command(apply_cli, None).await?;
         }
+        Some(Subcommand::ResponsesApiProxy(args)) => {
+            tokio::task::spawn_blocking(move || codex_responses_api_proxy::run_main(args))
+                .await??;
+        }
         Some(Subcommand::GenerateTs(gen_cli)) => {
             codex_protocol_ts::generate_ts(&gen_cli.out_dir, gen_cli.prettier.as_deref())?;
         }
@@ -460,7 +535,7 @@ mod tests {
             lines,
             vec![
                 "Token usage: total=2 input=0 output=2".to_string(),
-                "To continue this session, run codex resume 123e4567-e89b-12d3-a456-426614174000."
+                "To continue this session, run codex resume 123e4567-e89b-12d3-a456-426614174000"
                     .to_string(),
             ]
         );

codex-rs/cli/src/mcp_cmd.rs

@@ -1,6 +1,4 @@
-use std::collections::BTreeMap;
 use std::collections::HashMap;
-use std::path::PathBuf;
 
 use anyhow::Context;
 use anyhow::Result;
@@ -13,6 +11,7 @@ use codex_core::config::find_codex_home;
 use codex_core::config::load_global_mcp_servers;
 use codex_core::config::write_global_mcp_servers;
 use codex_core::config_types::McpServerConfig;
+use codex_core::config_types::McpServerTransportConfig;
 
 /// [experimental] Launch Codex as an MCP server or manage configured MCP servers.
 ///
@@ -28,14 +27,11 @@ pub struct McpCli {
     pub config_overrides: CliConfigOverrides,
 
     #[command(subcommand)]
-    pub cmd: Option<McpSubcommand>,
+    pub subcommand: McpSubcommand,
 }
 
 #[derive(Debug, clap::Subcommand)]
 pub enum McpSubcommand {
-    /// [experimental] Run the Codex MCP server (stdio transport).
-    Serve,
-
     /// [experimental] List configured MCP servers.
     List(ListArgs),
 
@@ -87,17 +83,13 @@ pub struct RemoveArgs {
 }
 
 impl McpCli {
-    pub async fn run(self, codex_linux_sandbox_exe: Option<PathBuf>) -> Result<()> {
+    pub async fn run(self) -> Result<()> {
         let McpCli {
             config_overrides,
-            cmd,
+            subcommand,
         } = self;
-        let subcommand = cmd.unwrap_or(McpSubcommand::Serve);
 
         match subcommand {
-            McpSubcommand::Serve => {
-                codex_mcp_server::run_main(codex_linux_sandbox_exe, config_overrides).await?;
-            }
             McpSubcommand::List(args) => {
                 run_list(&config_overrides, args)?;
             }
@@ -145,9 +137,11 @@ fn run_add(config_overrides: &CliConfigOverrides, add_args: AddArgs) -> Result<(
         .with_context(|| format!("failed to load MCP servers from {}", codex_home.display()))?;
 
     let new_entry = McpServerConfig {
-        command: command_bin,
-        args: command_args,
-        env: env_map,
+        transport: McpServerTransportConfig::Stdio {
+            command: command_bin,
+            args: command_args,
+            env: env_map,
+        },
         startup_timeout_sec: None,
         tool_timeout_sec: None,
     };
@@ -201,16 +195,25 @@ fn run_list(config_overrides: &CliConfigOverrides, list_args: ListArgs) -> Resul
         let json_entries: Vec<_> = entries
             .into_iter()
             .map(|(name, cfg)| {
-                let env = cfg.env.as_ref().map(|env| {
-                    env.iter()
-                        .map(|(k, v)| (k.clone(), v.clone()))
-                        .collect::<BTreeMap<_, _>>()
-                });
+                let transport = match &cfg.transport {
+                    McpServerTransportConfig::Stdio { command, args, env } => serde_json::json!({
+                        "type": "stdio",
+                        "command": command,
+                        "args": args,
+                        "env": env,
+                    }),
+                    McpServerTransportConfig::StreamableHttp { url, bearer_token } => {
+                        serde_json::json!({
+                            "type": "streamable_http",
+                            "url": url,
+                            "bearer_token": bearer_token,
+                        })
+                    }
+                };
+
                 serde_json::json!({
                     "name": name,
-                    "command": cfg.command,
-                    "args": cfg.args,
-                    "env": env,
+                    "transport": transport,
                     "startup_timeout_sec": cfg
                         .startup_timeout_sec
                         .map(|timeout| timeout.as_secs_f64()),
@@ -230,62 +233,111 @@ fn run_list(config_overrides: &CliConfigOverrides, list_args: ListArgs) -> Resul
         return Ok(());
     }
 
-    let mut rows: Vec<[String; 4]> = Vec::new();
+    let mut stdio_rows: Vec<[String; 4]> = Vec::new();
+    let mut http_rows: Vec<[String; 3]> = Vec::new();
+
     for (name, cfg) in entries {
-        let args = if cfg.args.is_empty() {
-            "-".to_string()
-        } else {
-            cfg.args.join(" ")
-        };
-
-        let env = match cfg.env.as_ref() {
-            None => "-".to_string(),
-            Some(map) if map.is_empty() => "-".to_string(),
-            Some(map) => {
-                let mut pairs: Vec<_> = map.iter().collect();
-                pairs.sort_by(|(a, _), (b, _)| a.cmp(b));
-                pairs
-                    .into_iter()
-                    .map(|(k, v)| format!("{k}={v}"))
-                    .collect::<Vec<_>>()
-                    .join(", ")
+        match &cfg.transport {
+            McpServerTransportConfig::Stdio { command, args, env } => {
+                let args_display = if args.is_empty() {
+                    "-".to_string()
+                } else {
+                    args.join(" ")
+                };
+                let env_display = match env.as_ref() {
+                    None => "-".to_string(),
+                    Some(map) if map.is_empty() => "-".to_string(),
+                    Some(map) => {
+                        let mut pairs: Vec<_> = map.iter().collect();
+                        pairs.sort_by(|(a, _), (b, _)| a.cmp(b));
+                        pairs
+                            .into_iter()
+                            .map(|(k, v)| format!("{k}={v}"))
+                            .collect::<Vec<_>>()
+                            .join(", ")
+                    }
+                };
+                stdio_rows.push([name.clone(), command.clone(), args_display, env_display]);
+            }
+            McpServerTransportConfig::StreamableHttp { url, bearer_token } => {
+                let has_bearer = if bearer_token.is_some() {
+                    "True"
+                } else {
+                    "False"
+                };
+                http_rows.push([name.clone(), url.clone(), has_bearer.into()]);
             }
-        };
-
-        rows.push([name.clone(), cfg.command.clone(), args, env]);
-    }
-
-    let mut widths = ["Name".len(), "Command".len(), "Args".len(), "Env".len()];
-    for row in &rows {
-        for (i, cell) in row.iter().enumerate() {
-            widths[i] = widths[i].max(cell.len());
         }
     }
 
-    println!(
-        "{:<name_w$}  {:<cmd_w$}  {:<args_w$}  {:<env_w$}",
-        "Name",
-        "Command",
-        "Args",
-        "Env",
-        name_w = widths[0],
-        cmd_w = widths[1],
-        args_w = widths[2],
-        env_w = widths[3],
-    );
+    if !stdio_rows.is_empty() {
+        let mut widths = ["Name".len(), "Command".len(), "Args".len(), "Env".len()];
+        for row in &stdio_rows {
+            for (i, cell) in row.iter().enumerate() {
+                widths[i] = widths[i].max(cell.len());
+            }
+        }
 
-    for row in rows {
         println!(
             "{:<name_w$}  {:<cmd_w$}  {:<args_w$}  {:<env_w$}",
-            row[0],
-            row[1],
-            row[2],
-            row[3],
+            "Name",
+            "Command",
+            "Args",
+            "Env",
             name_w = widths[0],
             cmd_w = widths[1],
             args_w = widths[2],
             env_w = widths[3],
         );
+
+        for row in &stdio_rows {
+            println!(
+                "{:<name_w$}  {:<cmd_w$}  {:<args_w$}  {:<env_w$}",
+                row[0],
+                row[1],
+                row[2],
+                row[3],
+                name_w = widths[0],
+                cmd_w = widths[1],
+                args_w = widths[2],
+                env_w = widths[3],
+            );
+        }
+    }
+
+    if !stdio_rows.is_empty() && !http_rows.is_empty() {
+        println!();
+    }
+
+    if !http_rows.is_empty() {
+        let mut widths = ["Name".len(), "Url".len(), "Has Bearer Token".len()];
+        for row in &http_rows {
+            for (i, cell) in row.iter().enumerate() {
+                widths[i] = widths[i].max(cell.len());
+            }
+        }
+
+        println!(
+            "{:<name_w$}  {:<url_w$}  {:<token_w$}",
+            "Name",
+            "Url",
+            "Has Bearer Token",
+            name_w = widths[0],
+            url_w = widths[1],
+            token_w = widths[2],
+        );
+
+        for row in &http_rows {
+            println!(
+                "{:<name_w$}  {:<url_w$}  {:<token_w$}",
+                row[0],
+                row[1],
+                row[2],
+                name_w = widths[0],
+                url_w = widths[1],
+                token_w = widths[2],
+            );
+        }
     }
 
     Ok(())
@@ -301,16 +353,22 @@ fn run_get(config_overrides: &CliConfigOverrides, get_args: GetArgs) -> Result<(
     };
 
     if get_args.json {
-        let env = server.env.as_ref().map(|env| {
-            env.iter()
-                .map(|(k, v)| (k.clone(), v.clone()))
-                .collect::<BTreeMap<_, _>>()
-        });
+        let transport = match &server.transport {
+            McpServerTransportConfig::Stdio { command, args, env } => serde_json::json!({
+                "type": "stdio",
+                "command": command,
+                "args": args,
+                "env": env,
+            }),
+            McpServerTransportConfig::StreamableHttp { url, bearer_token } => serde_json::json!({
+                "type": "streamable_http",
+                "url": url,
+                "bearer_token": bearer_token,
+            }),
+        };
         let output = serde_json::to_string_pretty(&serde_json::json!({
             "name": get_args.name,
-            "command": server.command,
-            "args": server.args,
-            "env": env,
+            "transport": transport,
             "startup_timeout_sec": server
                 .startup_timeout_sec
                 .map(|timeout| timeout.as_secs_f64()),
@@ -323,27 +381,38 @@ fn run_get(config_overrides: &CliConfigOverrides, get_args: GetArgs) -> Result<(
     }
 
     println!("{}", get_args.name);
-    println!("  command: {}", server.command);
-    let args = if server.args.is_empty() {
-        "-".to_string()
-    } else {
-        server.args.join(" ")
-    };
-    println!("  args: {args}");
-    let env_display = match server.env.as_ref() {
-        None => "-".to_string(),
-        Some(map) if map.is_empty() => "-".to_string(),
-        Some(map) => {
-            let mut pairs: Vec<_> = map.iter().collect();
-            pairs.sort_by(|(a, _), (b, _)| a.cmp(b));
-            pairs
-                .into_iter()
-                .map(|(k, v)| format!("{k}={v}"))
-                .collect::<Vec<_>>()
-                .join(", ")
+    match &server.transport {
+        McpServerTransportConfig::Stdio { command, args, env } => {
+            println!("  transport: stdio");
+            println!("  command: {command}");
+            let args_display = if args.is_empty() {
+                "-".to_string()
+            } else {
+                args.join(" ")
+            };
+            println!("  args: {args_display}");
+            let env_display = match env.as_ref() {
+                None => "-".to_string(),
+                Some(map) if map.is_empty() => "-".to_string(),
+                Some(map) => {
+                    let mut pairs: Vec<_> = map.iter().collect();
+                    pairs.sort_by(|(a, _), (b, _)| a.cmp(b));
+                    pairs
+                        .into_iter()
+                        .map(|(k, v)| format!("{k}={v}"))
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                }
+            };
+            println!("  env: {env_display}");
         }
-    };
-    println!("  env: {env_display}");
+        McpServerTransportConfig::StreamableHttp { url, bearer_token } => {
+            println!("  transport: streamable_http");
+            println!("  url: {url}");
+            let bearer = bearer_token.as_deref().unwrap_or("-");
+            println!("  bearer_token: {bearer}");
+        }
+    }
     if let Some(timeout) = server.startup_timeout_sec {
         println!("  startup_timeout_sec: {}", timeout.as_secs_f64());
     }

codex-rs/cli/tests/mcp_add_remove.rs

@@ -2,6 +2,7 @@ use std::path::Path;
 
 use anyhow::Result;
 use codex_core::config::load_global_mcp_servers;
+use codex_core::config_types::McpServerTransportConfig;
 use predicates::str::contains;
 use pretty_assertions::assert_eq;
 use tempfile::TempDir;
@@ -26,9 +27,14 @@ fn add_and_remove_server_updates_global_config() -> Result<()> {
     let servers = load_global_mcp_servers(codex_home.path())?;
     assert_eq!(servers.len(), 1);
     let docs = servers.get("docs").expect("server should exist");
-    assert_eq!(docs.command, "echo");
-    assert_eq!(docs.args, vec!["hello".to_string()]);
-    assert!(docs.env.is_none());
+    match &docs.transport {
+        McpServerTransportConfig::Stdio { command, args, env } => {
+            assert_eq!(command, "echo");
+            assert_eq!(args, &vec!["hello".to_string()]);
+            assert!(env.is_none());
+        }
+        other => panic!("unexpected transport: {other:?}"),
+    }
 
     let mut remove_cmd = codex_command(codex_home.path())?;
     remove_cmd
@@ -76,7 +82,10 @@ fn add_with_env_preserves_key_order_and_values() -> Result<()> {
 
     let servers = load_global_mcp_servers(codex_home.path())?;
     let envy = servers.get("envy").expect("server should exist");
-    let env = envy.env.as_ref().expect("env should be present");
+    let env = match &envy.transport {
+        McpServerTransportConfig::Stdio { env: Some(env), .. } => env,
+        other => panic!("unexpected transport: {other:?}"),
+    };
 
     assert_eq!(env.len(), 2);
     assert_eq!(env.get("FOO"), Some(&"bar".to_string()));

codex-rs/cli/tests/mcp_list.rs

@@ -4,6 +4,7 @@ use anyhow::Result;
 use predicates::str::contains;
 use pretty_assertions::assert_eq;
 use serde_json::Value as JsonValue;
+use serde_json::json;
 use tempfile::TempDir;
 
 fn codex_command(codex_home: &Path) -> Result<assert_cmd::Command> {
@@ -58,38 +59,35 @@ fn list_and_get_render_expected_output() -> Result<()> {
     assert!(json_output.status.success());
     let stdout = String::from_utf8(json_output.stdout)?;
     let parsed: JsonValue = serde_json::from_str(&stdout)?;
-    let array = parsed.as_array().expect("expected array");
-    assert_eq!(array.len(), 1);
-    let entry = &array[0];
-    assert_eq!(entry.get("name"), Some(&JsonValue::String("docs".into())));
     assert_eq!(
-        entry.get("command"),
-        Some(&JsonValue::String("docs-server".into()))
-    );
-
-    let args = entry
-        .get("args")
-        .and_then(|v| v.as_array())
-        .expect("args array");
-    assert_eq!(
-        args,
-        &vec![
-            JsonValue::String("--port".into()),
-            JsonValue::String("4000".into())
+        parsed,
+        json!([
+          {
+            "name": "docs",
+            "transport": {
+              "type": "stdio",
+              "command": "docs-server",
+              "args": [
+                "--port",
+                "4000"
+              ],
+              "env": {
+                "TOKEN": "secret"
+              }
+            },
+            "startup_timeout_sec": null,
+            "tool_timeout_sec": null
+          }
         ]
+        )
     );
 
-    let env = entry
-        .get("env")
-        .and_then(|v| v.as_object())
-        .expect("env map");
-    assert_eq!(env.get("TOKEN"), Some(&JsonValue::String("secret".into())));
-
     let mut get_cmd = codex_command(codex_home.path())?;
     let get_output = get_cmd.args(["mcp", "get", "docs"]).output()?;
     assert!(get_output.status.success());
     let stdout = String::from_utf8(get_output.stdout)?;
     assert!(stdout.contains("docs"));
+    assert!(stdout.contains("transport: stdio"));
     assert!(stdout.contains("command: docs-server"));
     assert!(stdout.contains("args: --port 4000"));
     assert!(stdout.contains("env: TOKEN=secret"));

codex-rs/cloud-tasks-client/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "codex-cloud-tasks-client"
+version = { workspace = true }
+edition = "2024"
+
+[lib]
+name = "codex_cloud_tasks_client"
+path = "src/lib.rs"
+
+[lints]
+workspace = true
+
+[features]
+default = ["online"]
+online = ["dep:codex-backend-client"]
+mock = []
+
+[dependencies]
+anyhow = "1"
+async-trait = "0.1"
+chrono = { version = "0.4", features = ["serde"] }
+diffy = "0.4.2"
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+thiserror = "2.0.12"
+codex-backend-client = { path = "../backend-client", optional = true }
+codex-git-apply = { path = "../git-apply" }

codex-rs/cloud-tasks-client/src/api.rs

@@ -0,0 +1,158 @@
+use chrono::DateTime;
+use chrono::Utc;
+use serde::Deserialize;
+use serde::Serialize;
+
+pub type Result<T> = std::result::Result<T, CloudTaskError>;
+
+#[derive(Debug, thiserror::Error)]
+pub enum CloudTaskError {
+    #[error("unimplemented: {0}")]
+    Unimplemented(&'static str),
+    #[error("http error: {0}")]
+    Http(String),
+    #[error("io error: {0}")]
+    Io(String),
+    #[error("{0}")]
+    Msg(String),
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct TaskId(pub String);
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum TaskStatus {
+    Pending,
+    Ready,
+    Applied,
+    Error,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TaskSummary {
+    pub id: TaskId,
+    pub title: String,
+    pub status: TaskStatus,
+    pub updated_at: DateTime<Utc>,
+    /// Backend environment identifier (when available)
+    pub environment_id: Option<String>,
+    /// Human-friendly environment label (when available)
+    pub environment_label: Option<String>,
+    pub summary: DiffSummary,
+    /// True when the backend reports this task as a code review.
+    #[serde(default)]
+    pub is_review: bool,
+    /// Number of assistant attempts (best-of-N), when reported by the backend.
+    #[serde(default)]
+    pub attempt_total: Option<usize>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
+pub enum AttemptStatus {
+    Pending,
+    InProgress,
+    Completed,
+    Failed,
+    Cancelled,
+    #[default]
+    Unknown,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TurnAttempt {
+    pub turn_id: String,
+    pub attempt_placement: Option<i64>,
+    pub created_at: Option<DateTime<Utc>>,
+    pub status: AttemptStatus,
+    pub diff: Option<String>,
+    pub messages: Vec<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ApplyStatus {
+    Success,
+    Partial,
+    Error,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ApplyOutcome {
+    pub applied: bool,
+    pub status: ApplyStatus,
+    pub message: String,
+    #[serde(default)]
+    pub skipped_paths: Vec<String>,
+    #[serde(default)]
+    pub conflict_paths: Vec<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct CreatedTask {
+    pub id: TaskId,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
+pub struct DiffSummary {
+    pub files_changed: usize,
+    pub lines_added: usize,
+    pub lines_removed: usize,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TaskText {
+    pub prompt: Option<String>,
+    pub messages: Vec<String>,
+    pub turn_id: Option<String>,
+    pub sibling_turn_ids: Vec<String>,
+    pub attempt_placement: Option<i64>,
+    pub attempt_status: AttemptStatus,
+}
+
+impl Default for TaskText {
+    fn default() -> Self {
+        Self {
+            prompt: None,
+            messages: Vec::new(),
+            turn_id: None,
+            sibling_turn_ids: Vec::new(),
+            attempt_placement: None,
+            attempt_status: AttemptStatus::Unknown,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+pub trait CloudBackend: Send + Sync {
+    async fn list_tasks(&self, env: Option<&str>) -> Result<Vec<TaskSummary>>;
+    async fn get_task_diff(&self, id: TaskId) -> Result<Option<String>>;
+    /// Return assistant output messages (no diff) when available.
+    async fn get_task_messages(&self, id: TaskId) -> Result<Vec<String>>;
+    /// Return the creating prompt and assistant messages (when available).
+    async fn get_task_text(&self, id: TaskId) -> Result<TaskText>;
+    /// Return any sibling attempts (best-of-N) for the given assistant turn.
+    async fn list_sibling_attempts(
+        &self,
+        task: TaskId,
+        turn_id: String,
+    ) -> Result<Vec<TurnAttempt>>;
+    /// Dry-run apply (preflight) that validates whether the patch would apply cleanly.
+    /// Never modifies the working tree. When `diff_override` is supplied, the provided diff is
+    /// used instead of re-fetching the task details so callers can apply alternate attempts.
+    async fn apply_task_preflight(
+        &self,
+        id: TaskId,
+        diff_override: Option<String>,
+    ) -> Result<ApplyOutcome>;
+    async fn apply_task(&self, id: TaskId, diff_override: Option<String>) -> Result<ApplyOutcome>;
+    async fn create_task(
+        &self,
+        env_id: &str,
+        prompt: &str,
+        git_ref: &str,
+        qa_mode: bool,
+        best_of_n: usize,
+    ) -> Result<CreatedTask>;
+}

codex-rs/cloud-tasks-client/src/http.rs

@@ -0,0 +1,769 @@
+use crate::ApplyOutcome;
+use crate::ApplyStatus;
+use crate::AttemptStatus;
+use crate::CloudBackend;
+use crate::CloudTaskError;
+use crate::DiffSummary;
+use crate::Result;
+use crate::TaskId;
+use crate::TaskStatus;
+use crate::TaskSummary;
+use crate::TurnAttempt;
+use crate::api::TaskText;
+use chrono::DateTime;
+use chrono::Utc;
+
+use codex_backend_client as backend;
+use codex_backend_client::CodeTaskDetailsResponseExt;
+
+#[derive(Clone)]
+pub struct HttpClient {
+    pub base_url: String,
+    backend: backend::Client,
+}
+
+impl HttpClient {
+    pub fn new(base_url: impl Into<String>) -> anyhow::Result<Self> {
+        let base_url = base_url.into();
+        let backend = backend::Client::new(base_url.clone())?;
+        Ok(Self { base_url, backend })
+    }
+
+    pub fn with_bearer_token(mut self, token: impl Into<String>) -> Self {
+        self.backend = self.backend.clone().with_bearer_token(token);
+        self
+    }
+
+    pub fn with_user_agent(mut self, ua: impl Into<String>) -> Self {
+        self.backend = self.backend.clone().with_user_agent(ua);
+        self
+    }
+
+    pub fn with_chatgpt_account_id(mut self, account_id: impl Into<String>) -> Self {
+        self.backend = self.backend.clone().with_chatgpt_account_id(account_id);
+        self
+    }
+
+    fn tasks_api(&self) -> api::Tasks<'_> {
+        api::Tasks::new(self)
+    }
+
+    fn attempts_api(&self) -> api::Attempts<'_> {
+        api::Attempts::new(self)
+    }
+
+    fn apply_api(&self) -> api::Apply<'_> {
+        api::Apply::new(self)
+    }
+}
+
+#[async_trait::async_trait]
+impl CloudBackend for HttpClient {
+    async fn list_tasks(&self, env: Option<&str>) -> Result<Vec<TaskSummary>> {
+        self.tasks_api().list(env).await
+    }
+
+    async fn get_task_diff(&self, id: TaskId) -> Result<Option<String>> {
+        self.tasks_api().diff(id).await
+    }
+
+    async fn get_task_messages(&self, id: TaskId) -> Result<Vec<String>> {
+        self.tasks_api().messages(id).await
+    }
+
+    async fn get_task_text(&self, id: TaskId) -> Result<TaskText> {
+        self.tasks_api().task_text(id).await
+    }
+
+    async fn list_sibling_attempts(
+        &self,
+        task: TaskId,
+        turn_id: String,
+    ) -> Result<Vec<TurnAttempt>> {
+        self.attempts_api().list(task, turn_id).await
+    }
+
+    async fn apply_task(&self, id: TaskId, diff_override: Option<String>) -> Result<ApplyOutcome> {
+        self.apply_api().run(id, diff_override, false).await
+    }
+
+    async fn apply_task_preflight(
+        &self,
+        id: TaskId,
+        diff_override: Option<String>,
+    ) -> Result<ApplyOutcome> {
+        self.apply_api().run(id, diff_override, true).await
+    }
+
+    async fn create_task(
+        &self,
+        env_id: &str,
+        prompt: &str,
+        git_ref: &str,
+        qa_mode: bool,
+        best_of_n: usize,
+    ) -> Result<crate::CreatedTask> {
+        self.tasks_api()
+            .create(env_id, prompt, git_ref, qa_mode, best_of_n)
+            .await
+    }
+}
+
+mod api {
+    use super::*;
+    use serde_json::Value;
+    use std::cmp::Ordering;
+    use std::collections::HashMap;
+
+    pub(crate) struct Tasks<'a> {
+        base_url: &'a str,
+        backend: &'a backend::Client,
+    }
+
+    impl<'a> Tasks<'a> {
+        pub(crate) fn new(client: &'a HttpClient) -> Self {
+            Self {
+                base_url: &client.base_url,
+                backend: &client.backend,
+            }
+        }
+
+        pub(crate) async fn list(&self, env: Option<&str>) -> Result<Vec<TaskSummary>> {
+            let resp = self
+                .backend
+                .list_tasks(Some(20), Some("current"), env)
+                .await
+                .map_err(|e| CloudTaskError::Http(format!("list_tasks failed: {e}")))?;
+
+            let tasks: Vec<TaskSummary> = resp
+                .items
+                .into_iter()
+                .map(map_task_list_item_to_summary)
+                .collect();
+
+            append_error_log(&format!(
+                "http.list_tasks: env={} items={}",
+                env.unwrap_or("<all>"),
+                tasks.len()
+            ));
+            Ok(tasks)
+        }
+
+        pub(crate) async fn diff(&self, id: TaskId) -> Result<Option<String>> {
+            let (details, body, ct) = self
+                .details_with_body(&id.0)
+                .await
+                .map_err(|e| CloudTaskError::Http(format!("get_task_details failed: {e}")))?;
+            if let Some(diff) = details.unified_diff() {
+                return Ok(Some(diff));
+            }
+            let _ = (body, ct);
+            Ok(None)
+        }
+
+        pub(crate) async fn messages(&self, id: TaskId) -> Result<Vec<String>> {
+            let (details, body, ct) = self
+                .details_with_body(&id.0)
+                .await
+                .map_err(|e| CloudTaskError::Http(format!("get_task_details failed: {e}")))?;
+
+            let mut msgs = details.assistant_text_messages();
+            if msgs.is_empty() {
+                msgs.extend(extract_assistant_messages_from_body(&body));
+            }
+            if !msgs.is_empty() {
+                return Ok(msgs);
+            }
+            if let Some(err) = details.assistant_error_message() {
+                return Ok(vec![format!("Task failed: {err}")]);
+            }
+
+            let url = match details_path(self.base_url, &id.0) {
+                Some(url) => url,
+                None => format!("{}/api/codex/tasks/{}", self.base_url, id.0),
+            };
+            Err(CloudTaskError::Http(format!(
+                "No assistant text messages in response. GET {url}; content-type={ct}; body={body}"
+            )))
+        }
+
+        pub(crate) async fn task_text(&self, id: TaskId) -> Result<TaskText> {
+            let (details, body, _ct) = self
+                .details_with_body(&id.0)
+                .await
+                .map_err(|e| CloudTaskError::Http(format!("get_task_details failed: {e}")))?;
+            let prompt = details.user_text_prompt();
+            let mut messages = details.assistant_text_messages();
+            if messages.is_empty() {
+                messages.extend(extract_assistant_messages_from_body(&body));
+            }
+            let assistant_turn = details.current_assistant_turn.as_ref();
+            let turn_id = assistant_turn.and_then(|turn| turn.id.clone());
+            let sibling_turn_ids = assistant_turn
+                .map(|turn| turn.sibling_turn_ids.clone())
+                .unwrap_or_default();
+            let attempt_placement = assistant_turn.and_then(|turn| turn.attempt_placement);
+            let attempt_status = attempt_status_from_str(
+                assistant_turn.and_then(|turn| turn.turn_status.as_deref()),
+            );
+            Ok(TaskText {
+                prompt,
+                messages,
+                turn_id,
+                sibling_turn_ids,
+                attempt_placement,
+                attempt_status,
+            })
+        }
+
+        pub(crate) async fn create(
+            &self,
+            env_id: &str,
+            prompt: &str,
+            git_ref: &str,
+            qa_mode: bool,
+            best_of_n: usize,
+        ) -> Result<crate::CreatedTask> {
+            let mut input_items: Vec<serde_json::Value> = Vec::new();
+            input_items.push(serde_json::json!({
+                "type": "message",
+                "role": "user",
+                "content": [{ "content_type": "text", "text": prompt }]
+            }));
+
+            if let Ok(diff) = std::env::var("CODEX_STARTING_DIFF")
+                && !diff.is_empty()
+            {
+                input_items.push(serde_json::json!({
+                    "type": "pre_apply_patch",
+                    "output_diff": { "diff": diff }
+                }));
+            }
+
+            let mut request_body = serde_json::json!({
+                "new_task": {
+                    "environment_id": env_id,
+                    "branch": git_ref,
+                    "run_environment_in_qa_mode": qa_mode,
+                },
+                "input_items": input_items,
+            });
+
+            if best_of_n > 1
+                && let Some(obj) = request_body.as_object_mut()
+            {
+                obj.insert(
+                    "metadata".to_string(),
+                    serde_json::json!({ "best_of_n": best_of_n }),
+                );
+            }
+
+            match self.backend.create_task(request_body).await {
+                Ok(id) => {
+                    append_error_log(&format!(
+                        "new_task: created id={id} env={} prompt_chars={}",
+                        env_id,
+                        prompt.chars().count()
+                    ));
+                    Ok(crate::CreatedTask { id: TaskId(id) })
+                }
+                Err(e) => {
+                    append_error_log(&format!(
+                        "new_task: create failed env={} prompt_chars={}: {}",
+                        env_id,
+                        prompt.chars().count(),
+                        e
+                    ));
+                    Err(CloudTaskError::Http(format!("create_task failed: {e}")))
+                }
+            }
+        }
+
+        async fn details_with_body(
+            &self,
+            id: &str,
+        ) -> anyhow::Result<(backend::CodeTaskDetailsResponse, String, String)> {
+            let (parsed, body, ct) = self.backend.get_task_details_with_body(id).await?;
+            Ok((parsed, body, ct))
+        }
+    }
+
+    pub(crate) struct Attempts<'a> {
+        backend: &'a backend::Client,
+    }
+
+    impl<'a> Attempts<'a> {
+        pub(crate) fn new(client: &'a HttpClient) -> Self {
+            Self {
+                backend: &client.backend,
+            }
+        }
+
+        pub(crate) async fn list(&self, task: TaskId, turn_id: String) -> Result<Vec<TurnAttempt>> {
+            let resp = self
+                .backend
+                .list_sibling_turns(&task.0, &turn_id)
+                .await
+                .map_err(|e| CloudTaskError::Http(format!("list_sibling_turns failed: {e}")))?;
+
+            let mut attempts: Vec<TurnAttempt> = resp
+                .sibling_turns
+                .iter()
+                .filter_map(turn_attempt_from_map)
+                .collect();
+            attempts.sort_by(compare_attempts);
+            Ok(attempts)
+        }
+    }
+
+    pub(crate) struct Apply<'a> {
+        backend: &'a backend::Client,
+    }
+
+    impl<'a> Apply<'a> {
+        pub(crate) fn new(client: &'a HttpClient) -> Self {
+            Self {
+                backend: &client.backend,
+            }
+        }
+
+        pub(crate) async fn run(
+            &self,
+            task_id: TaskId,
+            diff_override: Option<String>,
+            preflight: bool,
+        ) -> Result<ApplyOutcome> {
+            let id = task_id.0.clone();
+            let diff = match diff_override {
+                Some(diff) => diff,
+                None => {
+                    let details = self.backend.get_task_details(&id).await.map_err(|e| {
+                        CloudTaskError::Http(format!("get_task_details failed: {e}"))
+                    })?;
+                    details.unified_diff().ok_or_else(|| {
+                        CloudTaskError::Msg(format!("No diff available for task {id}"))
+                    })?
+                }
+            };
+
+            if !is_unified_diff(&diff) {
+                let summary = summarize_patch_for_logging(&diff);
+                let mode = if preflight { "preflight" } else { "apply" };
+                append_error_log(&format!(
+                    "apply_error: id={id} mode={mode} format=non-unified; {summary}"
+                ));
+                return Ok(ApplyOutcome {
+                    applied: false,
+                    status: ApplyStatus::Error,
+                    message: "Expected unified git diff; backend returned an incompatible format."
+                        .to_string(),
+                    skipped_paths: Vec::new(),
+                    conflict_paths: Vec::new(),
+                });
+            }
+
+            let req = codex_git_apply::ApplyGitRequest {
+                cwd: std::env::current_dir().unwrap_or_else(|_| std::env::temp_dir()),
+                diff: diff.clone(),
+                revert: false,
+                preflight,
+            };
+            let r = codex_git_apply::apply_git_patch(&req)
+                .map_err(|e| CloudTaskError::Io(format!("git apply failed to run: {e}")))?;
+
+            let status = if r.exit_code == 0 {
+                ApplyStatus::Success
+            } else if !r.applied_paths.is_empty() || !r.conflicted_paths.is_empty() {
+                ApplyStatus::Partial
+            } else {
+                ApplyStatus::Error
+            };
+            let applied = matches!(status, ApplyStatus::Success) && !preflight;
+
+            let message = if preflight {
+                match status {
+                    ApplyStatus::Success => {
+                        format!("Preflight passed for task {id} (applies cleanly)")
+                    }
+                    ApplyStatus::Partial => format!(
+                        "Preflight: patch does not fully apply for task {id} (applied={}, skipped={}, conflicts={})",
+                        r.applied_paths.len(),
+                        r.skipped_paths.len(),
+                        r.conflicted_paths.len()
+                    ),
+                    ApplyStatus::Error => format!(
+                        "Preflight failed for task {id} (applied={}, skipped={}, conflicts={})",
+                        r.applied_paths.len(),
+                        r.skipped_paths.len(),
+                        r.conflicted_paths.len()
+                    ),
+                }
+            } else {
+                match status {
+                    ApplyStatus::Success => format!(
+                        "Applied task {id} locally ({} files)",
+                        r.applied_paths.len()
+                    ),
+                    ApplyStatus::Partial => format!(
+                        "Apply partially succeeded for task {id} (applied={}, skipped={}, conflicts={})",
+                        r.applied_paths.len(),
+                        r.skipped_paths.len(),
+                        r.conflicted_paths.len()
+                    ),
+                    ApplyStatus::Error => format!(
+                        "Apply failed for task {id} (applied={}, skipped={}, conflicts={})",
+                        r.applied_paths.len(),
+                        r.skipped_paths.len(),
+                        r.conflicted_paths.len()
+                    ),
+                }
+            };
+
+            if matches!(status, ApplyStatus::Partial | ApplyStatus::Error)
+                || (preflight && !matches!(status, ApplyStatus::Success))
+            {
+                let mut log = String::new();
+                let summary = summarize_patch_for_logging(&diff);
+                let mode = if preflight { "preflight" } else { "apply" };
+                use std::fmt::Write as _;
+                let _ = writeln!(
+                    &mut log,
+                    "apply_result: mode={} id={} status={:?} applied={} skipped={} conflicts={} cmd={}",
+                    mode,
+                    id,
+                    status,
+                    r.applied_paths.len(),
+                    r.skipped_paths.len(),
+                    r.conflicted_paths.len(),
+                    r.cmd_for_log
+                );
+                let _ = writeln!(
+                    &mut log,
+                    "stdout_tail=\n{}\nstderr_tail=\n{}",
+                    tail(&r.stdout, 2000),
+                    tail(&r.stderr, 2000)
+                );
+                let _ = writeln!(&mut log, "{summary}");
+                let _ = writeln!(
+                    &mut log,
+                    "----- PATCH BEGIN -----\n{diff}\n----- PATCH END -----"
+                );
+                append_error_log(&log);
+            }
+
+            Ok(ApplyOutcome {
+                applied,
+                status,
+                message,
+                skipped_paths: r.skipped_paths,
+                conflict_paths: r.conflicted_paths,
+            })
+        }
+    }
+
+    fn details_path(base_url: &str, id: &str) -> Option<String> {
+        if base_url.contains("/backend-api") {
+            Some(format!("{base_url}/wham/tasks/{id}"))
+        } else if base_url.contains("/api/codex") {
+            Some(format!("{base_url}/tasks/{id}"))
+        } else {
+            None
+        }
+    }
+
+    fn extract_assistant_messages_from_body(body: &str) -> Vec<String> {
+        let mut msgs = Vec::new();
+        if let Ok(full) = serde_json::from_str::<serde_json::Value>(body)
+            && let Some(arr) = full
+                .get("current_assistant_turn")
+                .and_then(|v| v.get("worklog"))
+                .and_then(|v| v.get("messages"))
+                .and_then(|v| v.as_array())
+        {
+            for m in arr {
+                let is_assistant = m
+                    .get("author")
+                    .and_then(|a| a.get("role"))
+                    .and_then(|r| r.as_str())
+                    == Some("assistant");
+                if !is_assistant {
+                    continue;
+                }
+                if let Some(parts) = m
+                    .get("content")
+                    .and_then(|c| c.get("parts"))
+                    .and_then(|p| p.as_array())
+                {
+                    for p in parts {
+                        if let Some(s) = p.as_str() {
+                            if !s.is_empty() {
+                                msgs.push(s.to_string());
+                            }
+                            continue;
+                        }
+                        if let Some(obj) = p.as_object()
+                            && obj.get("content_type").and_then(|t| t.as_str()) == Some("text")
+                            && let Some(txt) = obj.get("text").and_then(|t| t.as_str())
+                        {
+                            msgs.push(txt.to_string());
+                        }
+                    }
+                }
+            }
+        }
+        msgs
+    }
+
+    fn turn_attempt_from_map(turn: &HashMap<String, Value>) -> Option<TurnAttempt> {
+        let turn_id = turn.get("id").and_then(Value::as_str)?.to_string();
+        let attempt_placement = turn.get("attempt_placement").and_then(Value::as_i64);
+        let created_at = parse_timestamp_value(turn.get("created_at"));
+        let status = attempt_status_from_str(turn.get("turn_status").and_then(Value::as_str));
+        let diff = extract_diff_from_turn(turn);
+        let messages = extract_assistant_messages_from_turn(turn);
+        Some(TurnAttempt {
+            turn_id,
+            attempt_placement,
+            created_at,
+            status,
+            diff,
+            messages,
+        })
+    }
+
+    fn compare_attempts(a: &TurnAttempt, b: &TurnAttempt) -> Ordering {
+        match (a.attempt_placement, b.attempt_placement) {
+            (Some(lhs), Some(rhs)) => lhs.cmp(&rhs),
+            (Some(_), None) => Ordering::Less,
+            (None, Some(_)) => Ordering::Greater,
+            (None, None) => match (a.created_at, b.created_at) {
+                (Some(lhs), Some(rhs)) => lhs.cmp(&rhs),
+                (Some(_), None) => Ordering::Less,
+                (None, Some(_)) => Ordering::Greater,
+                (None, None) => a.turn_id.cmp(&b.turn_id),
+            },
+        }
+    }
+
+    fn extract_diff_from_turn(turn: &HashMap<String, Value>) -> Option<String> {
+        let items = turn.get("output_items").and_then(Value::as_array)?;
+        for item in items {
+            match item.get("type").and_then(Value::as_str) {
+                Some("output_diff") => {
+                    if let Some(diff) = item.get("diff").and_then(Value::as_str)
+                        && !diff.is_empty()
+                    {
+                        return Some(diff.to_string());
+                    }
+                }
+                Some("pr") => {
+                    if let Some(diff) = item
+                        .get("output_diff")
+                        .and_then(Value::as_object)
+                        .and_then(|od| od.get("diff"))
+                        .and_then(Value::as_str)
+                        && !diff.is_empty()
+                    {
+                        return Some(diff.to_string());
+                    }
+                }
+                _ => {}
+            }
+        }
+        None
+    }
+
+    fn extract_assistant_messages_from_turn(turn: &HashMap<String, Value>) -> Vec<String> {
+        let mut msgs = Vec::new();
+        if let Some(items) = turn.get("output_items").and_then(Value::as_array) {
+            for item in items {
+                if item.get("type").and_then(Value::as_str) != Some("message") {
+                    continue;
+                }
+                if let Some(content) = item.get("content").and_then(Value::as_array) {
+                    for part in content {
+                        if part.get("content_type").and_then(Value::as_str) == Some("text")
+                            && let Some(txt) = part.get("text").and_then(Value::as_str)
+                            && !txt.is_empty()
+                        {
+                            msgs.push(txt.to_string());
+                        }
+                    }
+                }
+            }
+        }
+        msgs
+    }
+
+    fn attempt_status_from_str(raw: Option<&str>) -> AttemptStatus {
+        match raw.unwrap_or_default() {
+            "failed" => AttemptStatus::Failed,
+            "completed" => AttemptStatus::Completed,
+            "in_progress" => AttemptStatus::InProgress,
+            "pending" => AttemptStatus::Pending,
+            _ => AttemptStatus::Pending,
+        }
+    }
+
+    fn parse_timestamp_value(v: Option<&Value>) -> Option<DateTime<Utc>> {
+        let ts = v?.as_f64()?;
+        let secs = ts as i64;
+        let nanos = ((ts - secs as f64) * 1_000_000_000.0) as u32;
+        Some(DateTime::<Utc>::from(
+            std::time::UNIX_EPOCH + std::time::Duration::new(secs.max(0) as u64, nanos),
+        ))
+    }
+
+    fn map_task_list_item_to_summary(src: backend::TaskListItem) -> TaskSummary {
+        let status_display = src.task_status_display.as_ref();
+        TaskSummary {
+            id: TaskId(src.id),
+            title: src.title,
+            status: map_status(status_display),
+            updated_at: parse_updated_at(src.updated_at.as_ref()),
+            environment_id: None,
+            environment_label: env_label_from_status_display(status_display),
+            summary: diff_summary_from_status_display(status_display),
+            is_review: src
+                .pull_requests
+                .as_ref()
+                .is_some_and(|prs| !prs.is_empty()),
+            attempt_total: attempt_total_from_status_display(status_display),
+        }
+    }
+
+    fn map_status(v: Option<&HashMap<String, Value>>) -> TaskStatus {
+        if let Some(val) = v {
+            if let Some(turn) = val
+                .get("latest_turn_status_display")
+                .and_then(Value::as_object)
+                && let Some(s) = turn.get("turn_status").and_then(Value::as_str)
+            {
+                return match s {
+                    "failed" => TaskStatus::Error,
+                    "completed" => TaskStatus::Ready,
+                    "in_progress" => TaskStatus::Pending,
+                    "pending" => TaskStatus::Pending,
+                    "cancelled" => TaskStatus::Error,
+                    _ => TaskStatus::Pending,
+                };
+            }
+            if let Some(state) = val.get("state").and_then(Value::as_str) {
+                return match state {
+                    "pending" => TaskStatus::Pending,
+                    "ready" => TaskStatus::Ready,
+                    "applied" => TaskStatus::Applied,
+                    "error" => TaskStatus::Error,
+                    _ => TaskStatus::Pending,
+                };
+            }
+        }
+        TaskStatus::Pending
+    }
+
+    fn parse_updated_at(ts: Option<&f64>) -> DateTime<Utc> {
+        if let Some(v) = ts {
+            let secs = *v as i64;
+            let nanos = ((*v - secs as f64) * 1_000_000_000.0) as u32;
+            return DateTime::<Utc>::from(
+                std::time::UNIX_EPOCH + std::time::Duration::new(secs.max(0) as u64, nanos),
+            );
+        }
+        Utc::now()
+    }
+
+    fn env_label_from_status_display(v: Option<&HashMap<String, Value>>) -> Option<String> {
+        let map = v?;
+        map.get("environment_label")
+            .and_then(Value::as_str)
+            .map(str::to_string)
+    }
+
+    fn diff_summary_from_status_display(v: Option<&HashMap<String, Value>>) -> DiffSummary {
+        let mut out = DiffSummary::default();
+        let Some(map) = v else { return out };
+        let latest = map
+            .get("latest_turn_status_display")
+            .and_then(Value::as_object);
+        let Some(latest) = latest else { return out };
+        if let Some(ds) = latest.get("diff_stats").and_then(Value::as_object) {
+            if let Some(n) = ds.get("files_modified").and_then(Value::as_i64) {
+                out.files_changed = n.max(0) as usize;
+            }
+            if let Some(n) = ds.get("lines_added").and_then(Value::as_i64) {
+                out.lines_added = n.max(0) as usize;
+            }
+            if let Some(n) = ds.get("lines_removed").and_then(Value::as_i64) {
+                out.lines_removed = n.max(0) as usize;
+            }
+        }
+        out
+    }
+
+    fn attempt_total_from_status_display(v: Option<&HashMap<String, Value>>) -> Option<usize> {
+        let map = v?;
+        let latest = map
+            .get("latest_turn_status_display")
+            .and_then(Value::as_object)?;
+        let siblings = latest.get("sibling_turn_ids").and_then(Value::as_array)?;
+        Some(siblings.len().saturating_add(1))
+    }
+
+    fn is_unified_diff(diff: &str) -> bool {
+        let t = diff.trim_start();
+        if t.starts_with("diff --git ") {
+            return true;
+        }
+        let has_dash_headers = diff.contains("\n--- ") && diff.contains("\n+++ ");
+        let has_hunk = diff.contains("\n@@ ") || diff.starts_with("@@ ");
+        has_dash_headers && has_hunk
+    }
+
+    fn tail(s: &str, max: usize) -> String {
+        if s.len() <= max {
+            s.to_string()
+        } else {
+            s[s.len() - max..].to_string()
+        }
+    }
+
+    fn summarize_patch_for_logging(patch: &str) -> String {
+        let trimmed = patch.trim_start();
+        let kind = if trimmed.starts_with("*** Begin Patch") {
+            "codex-patch"
+        } else if trimmed.starts_with("diff --git ") || trimmed.contains("\n*** End Patch\n") {
+            "git-diff"
+        } else if trimmed.starts_with("@@ ") || trimmed.contains("\n@@ ") {
+            "unified-diff"
+        } else {
+            "unknown"
+        };
+        let lines = patch.lines().count();
+        let chars = patch.len();
+        let cwd = std::env::current_dir()
+            .ok()
+            .map(|p| p.display().to_string())
+            .unwrap_or_else(|| "<unknown>".to_string());
+        let head: String = patch.lines().take(20).collect::<Vec<&str>>().join("\n");
+        let head_trunc = if head.len() > 800 {
+            format!("{}…", &head[..800])
+        } else {
+            head
+        };
+        format!(
+            "patch_summary: kind={kind} lines={lines} chars={chars} cwd={cwd} ; head=\n{head_trunc}"
+        )
+    }
+}
+
+fn append_error_log(message: &str) {
+    let ts = Utc::now().to_rfc3339();
+    if let Ok(mut f) = std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open("error.log")
+    {
+        use std::io::Write as _;
+        let _ = writeln!(f, "[{ts}] {message}");
+    }
+}

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

@@ -0,0 +1,29 @@
+mod api;
+
+pub use api::ApplyOutcome;
+pub use api::ApplyStatus;
+pub use api::AttemptStatus;
+pub use api::CloudBackend;
+pub use api::CloudTaskError;
+pub use api::CreatedTask;
+pub use api::DiffSummary;
+pub use api::Result;
+pub use api::TaskId;
+pub use api::TaskStatus;
+pub use api::TaskSummary;
+pub use api::TaskText;
+pub use api::TurnAttempt;
+
+#[cfg(feature = "mock")]
+mod mock;
+
+#[cfg(feature = "online")]
+mod http;
+
+#[cfg(feature = "mock")]
+pub use mock::MockClient;
+
+#[cfg(feature = "online")]
+pub use http::HttpClient;
+
+// Reusable apply engine now lives in the shared crate `codex-git-apply`.

codex-rs/cloud-tasks-client/src/mock.rs

@@ -0,0 +1,180 @@
+use crate::ApplyOutcome;
+use crate::AttemptStatus;
+use crate::CloudBackend;
+use crate::DiffSummary;
+use crate::Result;
+use crate::TaskId;
+use crate::TaskStatus;
+use crate::TaskSummary;
+use crate::TurnAttempt;
+use crate::api::TaskText;
+use chrono::Utc;
+
+#[derive(Clone, Default)]
+pub struct MockClient;
+
+#[async_trait::async_trait]
+impl CloudBackend for MockClient {
+    async fn list_tasks(&self, _env: Option<&str>) -> Result<Vec<TaskSummary>> {
+        // Slightly vary content by env to aid tests that rely on the mock
+        let rows = match _env {
+            Some("env-A") => vec![("T-2000", "A: First", TaskStatus::Ready)],
+            Some("env-B") => vec![
+                ("T-3000", "B: One", TaskStatus::Ready),
+                ("T-3001", "B: Two", TaskStatus::Pending),
+            ],
+            _ => vec![
+                ("T-1000", "Update README formatting", TaskStatus::Ready),
+                ("T-1001", "Fix clippy warnings in core", TaskStatus::Pending),
+                ("T-1002", "Add contributing guide", TaskStatus::Ready),
+            ],
+        };
+        let environment_id = _env.map(str::to_string);
+        let environment_label = match _env {
+            Some("env-A") => Some("Env A".to_string()),
+            Some("env-B") => Some("Env B".to_string()),
+            Some(other) => Some(other.to_string()),
+            None => Some("Global".to_string()),
+        };
+        let mut out = Vec::new();
+        for (id_str, title, status) in rows {
+            let id = TaskId(id_str.to_string());
+            let diff = mock_diff_for(&id);
+            let (a, d) = count_from_unified(&diff);
+            out.push(TaskSummary {
+                id,
+                title: title.to_string(),
+                status,
+                updated_at: Utc::now(),
+                environment_id: environment_id.clone(),
+                environment_label: environment_label.clone(),
+                summary: DiffSummary {
+                    files_changed: 1,
+                    lines_added: a,
+                    lines_removed: d,
+                },
+                is_review: false,
+                attempt_total: Some(if id_str == "T-1000" { 2 } else { 1 }),
+            });
+        }
+        Ok(out)
+    }
+
+    async fn get_task_diff(&self, id: TaskId) -> Result<Option<String>> {
+        Ok(Some(mock_diff_for(&id)))
+    }
+
+    async fn get_task_messages(&self, _id: TaskId) -> Result<Vec<String>> {
+        Ok(vec![
+            "Mock assistant output: this task contains no diff.".to_string(),
+        ])
+    }
+
+    async fn get_task_text(&self, _id: TaskId) -> Result<TaskText> {
+        Ok(TaskText {
+            prompt: Some("Why is there no diff?".to_string()),
+            messages: vec!["Mock assistant output: this task contains no diff.".to_string()],
+            turn_id: Some("mock-turn".to_string()),
+            sibling_turn_ids: Vec::new(),
+            attempt_placement: Some(0),
+            attempt_status: AttemptStatus::Completed,
+        })
+    }
+
+    async fn apply_task(&self, id: TaskId, _diff_override: Option<String>) -> Result<ApplyOutcome> {
+        Ok(ApplyOutcome {
+            applied: true,
+            status: crate::ApplyStatus::Success,
+            message: format!("Applied task {} locally (mock)", id.0),
+            skipped_paths: Vec::new(),
+            conflict_paths: Vec::new(),
+        })
+    }
+
+    async fn apply_task_preflight(
+        &self,
+        id: TaskId,
+        _diff_override: Option<String>,
+    ) -> Result<ApplyOutcome> {
+        Ok(ApplyOutcome {
+            applied: false,
+            status: crate::ApplyStatus::Success,
+            message: format!("Preflight passed for task {} (mock)", id.0),
+            skipped_paths: Vec::new(),
+            conflict_paths: Vec::new(),
+        })
+    }
+
+    async fn list_sibling_attempts(
+        &self,
+        task: TaskId,
+        _turn_id: String,
+    ) -> Result<Vec<TurnAttempt>> {
+        if task.0 == "T-1000" {
+            return Ok(vec![TurnAttempt {
+                turn_id: "T-1000-attempt-2".to_string(),
+                attempt_placement: Some(1),
+                created_at: Some(Utc::now()),
+                status: AttemptStatus::Completed,
+                diff: Some(mock_diff_for(&task)),
+                messages: vec!["Mock alternate attempt".to_string()],
+            }]);
+        }
+        Ok(Vec::new())
+    }
+
+    async fn create_task(
+        &self,
+        env_id: &str,
+        prompt: &str,
+        git_ref: &str,
+        qa_mode: bool,
+        best_of_n: usize,
+    ) -> Result<crate::CreatedTask> {
+        let _ = (env_id, prompt, git_ref, qa_mode, best_of_n);
+        let id = format!("task_local_{}", chrono::Utc::now().timestamp_millis());
+        Ok(crate::CreatedTask { id: TaskId(id) })
+    }
+}
+
+fn mock_diff_for(id: &TaskId) -> String {
+    match id.0.as_str() {
+        "T-1000" => {
+            "diff --git a/README.md b/README.md\nindex 000000..111111 100644\n--- a/README.md\n+++ b/README.md\n@@ -1,2 +1,3 @@\n Intro\n-Hello\n+Hello, world!\n+Task: T-1000\n".to_string()
+        }
+        "T-1001" => {
+            "diff --git a/core/src/lib.rs b/core/src/lib.rs\nindex 000000..111111 100644\n--- a/core/src/lib.rs\n+++ b/core/src/lib.rs\n@@ -1,2 +1,1 @@\n-use foo;\n use bar;\n".to_string()
+        }
+        _ => {
+            "diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md\nindex 000000..111111 100644\n--- /dev/null\n+++ b/CONTRIBUTING.md\n@@ -0,0 +1,3 @@\n+## Contributing\n+Please open PRs.\n+Thanks!\n".to_string()
+        }
+    }
+}
+
+fn count_from_unified(diff: &str) -> (usize, usize) {
+    if let Ok(patch) = diffy::Patch::from_str(diff) {
+        patch
+            .hunks()
+            .iter()
+            .flat_map(diffy::Hunk::lines)
+            .fold((0, 0), |(a, d), l| match l {
+                diffy::Line::Insert(_) => (a + 1, d),
+                diffy::Line::Delete(_) => (a, d + 1),
+                _ => (a, d),
+            })
+    } else {
+        let mut a = 0;
+        let mut d = 0;
+        for l in diff.lines() {
+            if l.starts_with("+++") || l.starts_with("---") || l.starts_with("@@") {
+                continue;
+            }
+            match l.as_bytes().first() {
+                Some(b'+') => a += 1,
+                Some(b'-') => d += 1,
+                _ => {}
+            }
+        }
+        (a, d)
+    }
+}

codex-rs/cloud-tasks/Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "codex-cloud-tasks"
+version = { workspace = true }
+edition = "2024"
+
+[lib]
+name = "codex_cloud_tasks"
+path = "src/lib.rs"
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow = "1"
+clap = { version = "4", features = ["derive"] }
+codex-common = { path = "../common", features = ["cli"] }
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+tracing = { version = "0.1.41", features = ["log"] }
+tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
+codex-cloud-tasks-client = { path = "../cloud-tasks-client", features = ["mock", "online"] }
+ratatui = { version = "0.29.0" }
+crossterm = { version = "0.28.1", features = ["event-stream"] }
+tokio-stream = "0.1.17"
+chrono = { version = "0.4", features = ["serde"] }
+codex-login = { path = "../login" }
+codex-core = { path = "../core" }
+throbber-widgets-tui = "0.8.0"
+base64 = "0.22"
+serde_json = "1"
+reqwest = { version = "0.12", features = ["json"] }
+serde = { version = "1", features = ["derive"] }
+unicode-width = "0.1"
+codex-tui = { path = "../tui" }
+
+[dev-dependencies]
+async-trait = "0.1"

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

@@ -0,0 +1,482 @@
+use std::time::Duration;
+
+// Environment filter data models for the TUI
+#[derive(Clone, Debug, Default)]
+pub struct EnvironmentRow {
+    pub id: String,
+    pub label: Option<String>,
+    pub is_pinned: bool,
+    pub repo_hints: Option<String>, // e.g., "openai/codex"
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct EnvModalState {
+    pub query: String,
+    pub selected: usize,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct BestOfModalState {
+    pub selected: usize,
+}
+
+#[derive(Clone, Debug, Copy, PartialEq, Eq)]
+pub enum ApplyResultLevel {
+    Success,
+    Partial,
+    Error,
+}
+
+#[derive(Clone, Debug)]
+pub struct ApplyModalState {
+    pub task_id: TaskId,
+    pub title: String,
+    pub result_message: Option<String>,
+    pub result_level: Option<ApplyResultLevel>,
+    pub skipped_paths: Vec<String>,
+    pub conflict_paths: Vec<String>,
+    pub diff_override: Option<String>,
+}
+
+use crate::scrollable_diff::ScrollableDiff;
+use codex_cloud_tasks_client::CloudBackend;
+use codex_cloud_tasks_client::TaskId;
+use codex_cloud_tasks_client::TaskSummary;
+use throbber_widgets_tui::ThrobberState;
+
+#[derive(Default)]
+pub struct App {
+    pub tasks: Vec<TaskSummary>,
+    pub selected: usize,
+    pub status: String,
+    pub diff_overlay: Option<DiffOverlay>,
+    pub throbber: ThrobberState,
+    pub refresh_inflight: bool,
+    pub details_inflight: bool,
+    // Environment filter state
+    pub env_filter: Option<String>,
+    pub env_modal: Option<EnvModalState>,
+    pub apply_modal: Option<ApplyModalState>,
+    pub best_of_modal: Option<BestOfModalState>,
+    pub environments: Vec<EnvironmentRow>,
+    pub env_last_loaded: Option<std::time::Instant>,
+    pub env_loading: bool,
+    pub env_error: Option<String>,
+    // New Task page
+    pub new_task: Option<crate::new_task::NewTaskPage>,
+    pub best_of_n: usize,
+    // Apply preflight spinner state
+    pub apply_preflight_inflight: bool,
+    // Apply action spinner state
+    pub apply_inflight: bool,
+    // Background enrichment coordination
+    pub list_generation: u64,
+    pub in_flight: std::collections::HashSet<String>,
+    // Background enrichment caches were planned; currently unused.
+}
+
+impl App {
+    pub fn new() -> Self {
+        Self {
+            tasks: Vec::new(),
+            selected: 0,
+            status: "Press r to refresh".to_string(),
+            diff_overlay: None,
+            throbber: ThrobberState::default(),
+            refresh_inflight: false,
+            details_inflight: false,
+            env_filter: None,
+            env_modal: None,
+            apply_modal: None,
+            best_of_modal: None,
+            environments: Vec::new(),
+            env_last_loaded: None,
+            env_loading: false,
+            env_error: None,
+            new_task: None,
+            best_of_n: 1,
+            apply_preflight_inflight: false,
+            apply_inflight: false,
+            list_generation: 0,
+            in_flight: std::collections::HashSet::new(),
+        }
+    }
+
+    pub fn next(&mut self) {
+        if self.tasks.is_empty() {
+            return;
+        }
+        self.selected = (self.selected + 1).min(self.tasks.len().saturating_sub(1));
+    }
+
+    pub fn prev(&mut self) {
+        if self.tasks.is_empty() {
+            return;
+        }
+        if self.selected > 0 {
+            self.selected -= 1;
+        }
+    }
+}
+
+pub async fn load_tasks(
+    backend: &dyn CloudBackend,
+    env: Option<&str>,
+) -> anyhow::Result<Vec<TaskSummary>> {
+    // In later milestones, add a small debounce, spinner, and error display.
+    let tasks = tokio::time::timeout(Duration::from_secs(5), backend.list_tasks(env)).await??;
+    // Hide review-only tasks from the main list.
+    let filtered: Vec<TaskSummary> = tasks.into_iter().filter(|t| !t.is_review).collect();
+    Ok(filtered)
+}
+
+pub struct DiffOverlay {
+    pub title: String,
+    pub task_id: TaskId,
+    pub sd: ScrollableDiff,
+    pub base_can_apply: bool,
+    pub diff_lines: Vec<String>,
+    pub text_lines: Vec<String>,
+    pub prompt: Option<String>,
+    pub attempts: Vec<AttemptView>,
+    pub selected_attempt: usize,
+    pub current_view: DetailView,
+    pub base_turn_id: Option<String>,
+    pub sibling_turn_ids: Vec<String>,
+    pub attempt_total_hint: Option<usize>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct AttemptView {
+    pub turn_id: Option<String>,
+    pub status: codex_cloud_tasks_client::AttemptStatus,
+    pub attempt_placement: Option<i64>,
+    pub diff_lines: Vec<String>,
+    pub text_lines: Vec<String>,
+    pub prompt: Option<String>,
+    pub diff_raw: Option<String>,
+}
+
+impl AttemptView {
+    pub fn has_diff(&self) -> bool {
+        !self.diff_lines.is_empty()
+    }
+
+    pub fn has_text(&self) -> bool {
+        !self.text_lines.is_empty() || self.prompt.is_some()
+    }
+}
+
+impl DiffOverlay {
+    pub fn new(task_id: TaskId, title: String, attempt_total_hint: Option<usize>) -> Self {
+        let mut sd = ScrollableDiff::new();
+        sd.set_content(Vec::new());
+        Self {
+            title,
+            task_id,
+            sd,
+            base_can_apply: false,
+            diff_lines: Vec::new(),
+            text_lines: Vec::new(),
+            prompt: None,
+            attempts: vec![AttemptView::default()],
+            selected_attempt: 0,
+            current_view: DetailView::Prompt,
+            base_turn_id: None,
+            sibling_turn_ids: Vec::new(),
+            attempt_total_hint,
+        }
+    }
+
+    pub fn current_attempt(&self) -> Option<&AttemptView> {
+        self.attempts.get(self.selected_attempt)
+    }
+
+    pub fn base_attempt_mut(&mut self) -> &mut AttemptView {
+        if self.attempts.is_empty() {
+            self.attempts.push(AttemptView::default());
+        }
+        &mut self.attempts[0]
+    }
+
+    pub fn set_view(&mut self, view: DetailView) {
+        self.current_view = view;
+        self.apply_selection_to_fields();
+    }
+
+    pub fn expected_attempts(&self) -> Option<usize> {
+        self.attempt_total_hint.or({
+            if self.attempts.is_empty() {
+                None
+            } else {
+                Some(self.attempts.len())
+            }
+        })
+    }
+
+    pub fn attempt_count(&self) -> usize {
+        self.attempts.len()
+    }
+
+    pub fn attempt_display_total(&self) -> usize {
+        self.expected_attempts()
+            .unwrap_or_else(|| self.attempts.len().max(1))
+    }
+
+    pub fn step_attempt(&mut self, delta: isize) -> bool {
+        let total = self.attempts.len();
+        if total <= 1 {
+            return false;
+        }
+        let total_isize = total as isize;
+        let current = self.selected_attempt as isize;
+        let mut next = current + delta;
+        next = ((next % total_isize) + total_isize) % total_isize;
+        let next = next as usize;
+        self.selected_attempt = next;
+        self.apply_selection_to_fields();
+        true
+    }
+
+    pub fn current_can_apply(&self) -> bool {
+        matches!(self.current_view, DetailView::Diff)
+            && self
+                .current_attempt()
+                .and_then(|attempt| attempt.diff_raw.as_ref())
+                .map(|diff| !diff.is_empty())
+                .unwrap_or(false)
+    }
+
+    pub fn apply_selection_to_fields(&mut self) {
+        let (diff_lines, text_lines, prompt) = if let Some(attempt) = self.current_attempt() {
+            (
+                attempt.diff_lines.clone(),
+                attempt.text_lines.clone(),
+                attempt.prompt.clone(),
+            )
+        } else {
+            self.diff_lines.clear();
+            self.text_lines.clear();
+            self.prompt = None;
+            self.sd.set_content(vec!["<loading attempt>".to_string()]);
+            return;
+        };
+
+        self.diff_lines = diff_lines.clone();
+        self.text_lines = text_lines.clone();
+        self.prompt = prompt;
+
+        match self.current_view {
+            DetailView::Diff => {
+                if diff_lines.is_empty() {
+                    self.sd.set_content(vec!["<no diff available>".to_string()]);
+                } else {
+                    self.sd.set_content(diff_lines);
+                }
+            }
+            DetailView::Prompt => {
+                if text_lines.is_empty() {
+                    self.sd.set_content(vec!["<no output>".to_string()]);
+                } else {
+                    self.sd.set_content(text_lines);
+                }
+            }
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum DetailView {
+    Diff,
+    Prompt,
+}
+
+/// Internal app events delivered from background tasks.
+/// These let the UI event loop remain responsive and keep the spinner animating.
+#[derive(Debug)]
+pub enum AppEvent {
+    TasksLoaded {
+        env: Option<String>,
+        result: anyhow::Result<Vec<TaskSummary>>,
+    },
+    // Background diff summary events were planned; removed for now to keep code minimal.
+    /// Autodetection of a likely environment id finished
+    EnvironmentAutodetected(anyhow::Result<crate::env_detect::AutodetectSelection>),
+    /// Background completion of environment list fetch
+    EnvironmentsLoaded(anyhow::Result<Vec<EnvironmentRow>>),
+    DetailsDiffLoaded {
+        id: TaskId,
+        title: String,
+        diff: String,
+    },
+    DetailsMessagesLoaded {
+        id: TaskId,
+        title: String,
+        messages: Vec<String>,
+        prompt: Option<String>,
+        turn_id: Option<String>,
+        sibling_turn_ids: Vec<String>,
+        attempt_placement: Option<i64>,
+        attempt_status: codex_cloud_tasks_client::AttemptStatus,
+    },
+    DetailsFailed {
+        id: TaskId,
+        title: String,
+        error: String,
+    },
+    AttemptsLoaded {
+        id: TaskId,
+        attempts: Vec<codex_cloud_tasks_client::TurnAttempt>,
+    },
+    /// Background completion of new task submission
+    NewTaskSubmitted(Result<codex_cloud_tasks_client::CreatedTask, String>),
+    /// Background completion of apply preflight when opening modal or on demand
+    ApplyPreflightFinished {
+        id: TaskId,
+        title: String,
+        message: String,
+        level: ApplyResultLevel,
+        skipped: Vec<String>,
+        conflicts: Vec<String>,
+    },
+    /// Background completion of apply action (actual patch application)
+    ApplyFinished {
+        id: TaskId,
+        result: std::result::Result<codex_cloud_tasks_client::ApplyOutcome, String>,
+    },
+}
+
+// Convenience aliases; currently unused.
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+
+    struct FakeBackend {
+        // maps env key to titles
+        by_env: std::collections::HashMap<Option<String>, Vec<&'static str>>,
+    }
+
+    #[async_trait::async_trait]
+    impl codex_cloud_tasks_client::CloudBackend for FakeBackend {
+        async fn list_tasks(
+            &self,
+            env: Option<&str>,
+        ) -> codex_cloud_tasks_client::Result<Vec<TaskSummary>> {
+            let key = env.map(str::to_string);
+            let titles = self
+                .by_env
+                .get(&key)
+                .cloned()
+                .unwrap_or_else(|| vec!["default-a", "default-b"]);
+            let mut out = Vec::new();
+            for (i, t) in titles.into_iter().enumerate() {
+                out.push(TaskSummary {
+                    id: TaskId(format!("T-{i}")),
+                    title: t.to_string(),
+                    status: codex_cloud_tasks_client::TaskStatus::Ready,
+                    updated_at: Utc::now(),
+                    environment_id: env.map(str::to_string),
+                    environment_label: None,
+                    summary: codex_cloud_tasks_client::DiffSummary::default(),
+                    is_review: false,
+                    attempt_total: Some(1),
+                });
+            }
+            Ok(out)
+        }
+
+        async fn get_task_diff(
+            &self,
+            _id: TaskId,
+        ) -> codex_cloud_tasks_client::Result<Option<String>> {
+            Err(codex_cloud_tasks_client::CloudTaskError::Unimplemented(
+                "not used in test",
+            ))
+        }
+
+        async fn get_task_messages(
+            &self,
+            _id: TaskId,
+        ) -> codex_cloud_tasks_client::Result<Vec<String>> {
+            Ok(vec![])
+        }
+        async fn get_task_text(
+            &self,
+            _id: TaskId,
+        ) -> codex_cloud_tasks_client::Result<codex_cloud_tasks_client::TaskText> {
+            Ok(codex_cloud_tasks_client::TaskText {
+                prompt: Some("Example prompt".to_string()),
+                messages: Vec::new(),
+                turn_id: Some("fake-turn".to_string()),
+                sibling_turn_ids: Vec::new(),
+                attempt_placement: Some(0),
+                attempt_status: codex_cloud_tasks_client::AttemptStatus::Completed,
+            })
+        }
+
+        async fn list_sibling_attempts(
+            &self,
+            _task: TaskId,
+            _turn_id: String,
+        ) -> codex_cloud_tasks_client::Result<Vec<codex_cloud_tasks_client::TurnAttempt>> {
+            Ok(Vec::new())
+        }
+
+        async fn apply_task(
+            &self,
+            _id: TaskId,
+            _diff_override: Option<String>,
+        ) -> codex_cloud_tasks_client::Result<codex_cloud_tasks_client::ApplyOutcome> {
+            Err(codex_cloud_tasks_client::CloudTaskError::Unimplemented(
+                "not used in test",
+            ))
+        }
+
+        async fn apply_task_preflight(
+            &self,
+            _id: TaskId,
+            _diff_override: Option<String>,
+        ) -> codex_cloud_tasks_client::Result<codex_cloud_tasks_client::ApplyOutcome> {
+            Err(codex_cloud_tasks_client::CloudTaskError::Unimplemented(
+                "not used in test",
+            ))
+        }
+
+        async fn create_task(
+            &self,
+            _env_id: &str,
+            _prompt: &str,
+            _git_ref: &str,
+            _qa_mode: bool,
+            _best_of_n: usize,
+        ) -> codex_cloud_tasks_client::Result<codex_cloud_tasks_client::CreatedTask> {
+            Err(codex_cloud_tasks_client::CloudTaskError::Unimplemented(
+                "not used in test",
+            ))
+        }
+    }
+
+    #[tokio::test]
+    async fn load_tasks_uses_env_parameter() {
+        // Arrange: env-specific task titles
+        let mut by_env = std::collections::HashMap::new();
+        by_env.insert(None, vec!["root-1", "root-2"]);
+        by_env.insert(Some("env-A".to_string()), vec!["A-1"]);
+        by_env.insert(Some("env-B".to_string()), vec!["B-1", "B-2", "B-3"]);
+        let backend = FakeBackend { by_env };
+
+        // Act + Assert
+        let root = load_tasks(&backend, None).await.unwrap();
+        assert_eq!(root.len(), 2);
+        assert_eq!(root[0].title, "root-1");
+
+        let a = load_tasks(&backend, Some("env-A")).await.unwrap();
+        assert_eq!(a.len(), 1);
+        assert_eq!(a[0].title, "A-1");
+
+        let b = load_tasks(&backend, Some("env-B")).await.unwrap();
+        assert_eq!(b.len(), 3);
+        assert_eq!(b[2].title, "B-3");
+    }
+}

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

@@ -0,0 +1,9 @@
+use clap::Parser;
+use codex_common::CliConfigOverrides;
+
+#[derive(Parser, Debug, Default)]
+#[command(version)]
+pub struct Cli {
+    #[clap(skip)]
+    pub config_overrides: CliConfigOverrides,
+}

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

@@ -0,0 +1,361 @@
+use reqwest::header::CONTENT_TYPE;
+use reqwest::header::HeaderMap;
+use std::collections::HashMap;
+use tracing::info;
+use tracing::warn;
+
+#[derive(Debug, Clone, serde::Deserialize)]
+struct CodeEnvironment {
+    id: String,
+    #[serde(default)]
+    label: Option<String>,
+    #[serde(default)]
+    is_pinned: Option<bool>,
+    #[serde(default)]
+    task_count: Option<i64>,
+}
+
+#[derive(Debug, Clone)]
+pub struct AutodetectSelection {
+    pub id: String,
+    pub label: Option<String>,
+}
+
+pub async fn autodetect_environment_id(
+    base_url: &str,
+    headers: &HeaderMap,
+    desired_label: Option<String>,
+) -> anyhow::Result<AutodetectSelection> {
+    // 1) Try repo-specific environments based on local git origins (GitHub only, like VSCode)
+    let origins = get_git_origins();
+    crate::append_error_log(format!("env: git origins: {origins:?}"));
+    let mut by_repo_envs: Vec<CodeEnvironment> = Vec::new();
+    for origin in &origins {
+        if let Some((owner, repo)) = parse_owner_repo(origin) {
+            let url = if base_url.contains("/backend-api") {
+                format!(
+                    "{}/wham/environments/by-repo/{}/{}/{}",
+                    base_url, "github", owner, repo
+                )
+            } else {
+                format!(
+                    "{}/api/codex/environments/by-repo/{}/{}/{}",
+                    base_url, "github", owner, repo
+                )
+            };
+            crate::append_error_log(format!("env: GET {url}"));
+            match get_json::<Vec<CodeEnvironment>>(&url, headers).await {
+                Ok(mut list) => {
+                    crate::append_error_log(format!(
+                        "env: by-repo returned {} env(s) for {owner}/{repo}",
+                        list.len(),
+                    ));
+                    by_repo_envs.append(&mut list);
+                }
+                Err(e) => crate::append_error_log(format!(
+                    "env: by-repo fetch failed for {owner}/{repo}: {e}"
+                )),
+            }
+        }
+    }
+    if let Some(env) = pick_environment_row(&by_repo_envs, desired_label.as_deref()) {
+        return Ok(AutodetectSelection {
+            id: env.id.clone(),
+            label: env.label.as_deref().map(str::to_owned),
+        });
+    }
+
+    // 2) Fallback to the full list
+    let list_url = if base_url.contains("/backend-api") {
+        format!("{base_url}/wham/environments")
+    } else {
+        format!("{base_url}/api/codex/environments")
+    };
+    crate::append_error_log(format!("env: GET {list_url}"));
+    // Fetch and log the full environments JSON for debugging
+    let http = reqwest::Client::builder().build()?;
+    let res = http.get(&list_url).headers(headers.clone()).send().await?;
+    let status = res.status();
+    let ct = res
+        .headers()
+        .get(CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("")
+        .to_string();
+    let body = res.text().await.unwrap_or_default();
+    crate::append_error_log(format!("env: status={status} content-type={ct}"));
+    match serde_json::from_str::<serde_json::Value>(&body) {
+        Ok(v) => {
+            let pretty = serde_json::to_string_pretty(&v).unwrap_or(body.clone());
+            crate::append_error_log(format!("env: /environments JSON (pretty):\n{pretty}"));
+        }
+        Err(_) => crate::append_error_log(format!("env: /environments (raw):\n{body}")),
+    }
+    if !status.is_success() {
+        anyhow::bail!("GET {list_url} failed: {status}; content-type={ct}; body={body}");
+    }
+    let all_envs: Vec<CodeEnvironment> = serde_json::from_str(&body).map_err(|e| {
+        anyhow::anyhow!("Decode error for {list_url}: {e}; content-type={ct}; body={body}")
+    })?;
+    if let Some(env) = pick_environment_row(&all_envs, desired_label.as_deref()) {
+        return Ok(AutodetectSelection {
+            id: env.id.clone(),
+            label: env.label.as_deref().map(str::to_owned),
+        });
+    }
+    anyhow::bail!("no environments available")
+}
+
+fn pick_environment_row(
+    envs: &[CodeEnvironment],
+    desired_label: Option<&str>,
+) -> Option<CodeEnvironment> {
+    if envs.is_empty() {
+        return None;
+    }
+    if let Some(label) = desired_label {
+        let lc = label.to_lowercase();
+        if let Some(e) = envs
+            .iter()
+            .find(|e| e.label.as_deref().unwrap_or("").to_lowercase() == lc)
+        {
+            crate::append_error_log(format!("env: matched by label: {label} -> {}", e.id));
+            return Some(e.clone());
+        }
+    }
+    if envs.len() == 1 {
+        crate::append_error_log("env: single environment available; selecting it");
+        return Some(envs[0].clone());
+    }
+    if let Some(e) = envs.iter().find(|e| e.is_pinned.unwrap_or(false)) {
+        crate::append_error_log(format!("env: selecting pinned environment: {}", e.id));
+        return Some(e.clone());
+    }
+    // Highest task_count as heuristic
+    if let Some(e) = envs
+        .iter()
+        .max_by_key(|e| e.task_count.unwrap_or(0))
+        .or_else(|| envs.first())
+    {
+        crate::append_error_log(format!("env: selecting by task_count/first: {}", e.id));
+        return Some(e.clone());
+    }
+    None
+}
+
+async fn get_json<T: serde::de::DeserializeOwned>(
+    url: &str,
+    headers: &HeaderMap,
+) -> anyhow::Result<T> {
+    let http = reqwest::Client::builder().build()?;
+    let res = http.get(url).headers(headers.clone()).send().await?;
+    let status = res.status();
+    let ct = res
+        .headers()
+        .get(CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or("")
+        .to_string();
+    let body = res.text().await.unwrap_or_default();
+    crate::append_error_log(format!("env: status={status} content-type={ct}"));
+    if !status.is_success() {
+        anyhow::bail!("GET {url} failed: {status}; content-type={ct}; body={body}");
+    }
+    let parsed = serde_json::from_str::<T>(&body).map_err(|e| {
+        anyhow::anyhow!("Decode error for {url}: {e}; content-type={ct}; body={body}")
+    })?;
+    Ok(parsed)
+}
+
+fn get_git_origins() -> Vec<String> {
+    // Prefer: git config --get-regexp remote\..*\.url
+    let out = std::process::Command::new("git")
+        .args(["config", "--get-regexp", "remote\\..*\\.url"])
+        .output();
+    if let Ok(ok) = out
+        && ok.status.success()
+    {
+        let s = String::from_utf8_lossy(&ok.stdout);
+        let mut urls = Vec::new();
+        for line in s.lines() {
+            if let Some((_, url)) = line.split_once(' ') {
+                urls.push(url.trim().to_string());
+            }
+        }
+        if !urls.is_empty() {
+            return uniq(urls);
+        }
+    }
+    // Fallback: git remote -v
+    let out = std::process::Command::new("git")
+        .args(["remote", "-v"])
+        .output();
+    if let Ok(ok) = out
+        && ok.status.success()
+    {
+        let s = String::from_utf8_lossy(&ok.stdout);
+        let mut urls = Vec::new();
+        for line in s.lines() {
+            let parts: Vec<&str> = line.split_whitespace().collect();
+            if parts.len() >= 2 {
+                urls.push(parts[1].to_string());
+            }
+        }
+        if !urls.is_empty() {
+            return uniq(urls);
+        }
+    }
+    Vec::new()
+}
+
+fn uniq(mut v: Vec<String>) -> Vec<String> {
+    v.sort();
+    v.dedup();
+    v
+}
+
+fn parse_owner_repo(url: &str) -> Option<(String, String)> {
+    // Normalize common prefixes and handle multiple SSH/HTTPS variants.
+    let mut s = url.trim().to_string();
+    // Drop protocol scheme for ssh URLs
+    if let Some(rest) = s.strip_prefix("ssh://") {
+        s = rest.to_string();
+    }
+    // Accept any user before @github.com (e.g., git@, org-123@)
+    if let Some(idx) = s.find("@github.com:") {
+        let rest = &s[idx + "@github.com:".len()..];
+        let rest = rest.trim_start_matches('/').trim_end_matches(".git");
+        let mut parts = rest.splitn(2, '/');
+        let owner = parts.next()?.to_string();
+        let repo = parts.next()?.to_string();
+        crate::append_error_log(format!("env: parsed SSH GitHub origin => {owner}/{repo}"));
+        return Some((owner, repo));
+    }
+    // HTTPS or git protocol
+    for prefix in [
+        "https://github.com/",
+        "http://github.com/",
+        "git://github.com/",
+        "github.com/",
+    ] {
+        if let Some(rest) = s.strip_prefix(prefix) {
+            let rest = rest.trim_start_matches('/').trim_end_matches(".git");
+            let mut parts = rest.splitn(2, '/');
+            let owner = parts.next()?.to_string();
+            let repo = parts.next()?.to_string();
+            crate::append_error_log(format!("env: parsed HTTP GitHub origin => {owner}/{repo}"));
+            return Some((owner, repo));
+        }
+    }
+    None
+}
+
+/// List environments for the current repo(s) with a fallback to the global list.
+/// Returns a de-duplicated, sorted set suitable for the TUI modal.
+pub async fn list_environments(
+    base_url: &str,
+    headers: &HeaderMap,
+) -> anyhow::Result<Vec<crate::app::EnvironmentRow>> {
+    let mut map: HashMap<String, crate::app::EnvironmentRow> = HashMap::new();
+
+    // 1) By-repo lookup for each parsed GitHub origin
+    let origins = get_git_origins();
+    for origin in &origins {
+        if let Some((owner, repo)) = parse_owner_repo(origin) {
+            let url = if base_url.contains("/backend-api") {
+                format!(
+                    "{}/wham/environments/by-repo/{}/{}/{}",
+                    base_url, "github", owner, repo
+                )
+            } else {
+                format!(
+                    "{}/api/codex/environments/by-repo/{}/{}/{}",
+                    base_url, "github", owner, repo
+                )
+            };
+            match get_json::<Vec<CodeEnvironment>>(&url, headers).await {
+                Ok(list) => {
+                    info!("env_tui: by-repo {}:{} -> {} envs", owner, repo, list.len());
+                    for e in list {
+                        let entry =
+                            map.entry(e.id.clone())
+                                .or_insert_with(|| crate::app::EnvironmentRow {
+                                    id: e.id.clone(),
+                                    label: e.label.clone(),
+                                    is_pinned: e.is_pinned.unwrap_or(false),
+                                    repo_hints: Some(format!("{owner}/{repo}")),
+                                });
+                        // Merge: keep label if present, or use new; accumulate pinned flag
+                        if entry.label.is_none() {
+                            entry.label = e.label.clone();
+                        }
+                        entry.is_pinned = entry.is_pinned || e.is_pinned.unwrap_or(false);
+                        if entry.repo_hints.is_none() {
+                            entry.repo_hints = Some(format!("{owner}/{repo}"));
+                        }
+                    }
+                }
+                Err(e) => {
+                    warn!(
+                        "env_tui: by-repo fetch failed for {}/{}: {}",
+                        owner, repo, e
+                    );
+                }
+            }
+        }
+    }
+
+    // 2) Fallback to the full list; on error return what we have if any.
+    let list_url = if base_url.contains("/backend-api") {
+        format!("{base_url}/wham/environments")
+    } else {
+        format!("{base_url}/api/codex/environments")
+    };
+    match get_json::<Vec<CodeEnvironment>>(&list_url, headers).await {
+        Ok(list) => {
+            info!("env_tui: global list -> {} envs", list.len());
+            for e in list {
+                let entry = map
+                    .entry(e.id.clone())
+                    .or_insert_with(|| crate::app::EnvironmentRow {
+                        id: e.id.clone(),
+                        label: e.label.clone(),
+                        is_pinned: e.is_pinned.unwrap_or(false),
+                        repo_hints: None,
+                    });
+                if entry.label.is_none() {
+                    entry.label = e.label.clone();
+                }
+                entry.is_pinned = entry.is_pinned || e.is_pinned.unwrap_or(false);
+            }
+        }
+        Err(e) => {
+            if map.is_empty() {
+                return Err(e);
+            } else {
+                warn!(
+                    "env_tui: global list failed; using by-repo results only: {}",
+                    e
+                );
+            }
+        }
+    }
+
+    let mut rows: Vec<crate::app::EnvironmentRow> = map.into_values().collect();
+    rows.sort_by(|a, b| {
+        // pinned first
+        let p = b.is_pinned.cmp(&a.is_pinned);
+        if p != std::cmp::Ordering::Equal {
+            return p;
+        }
+        // then label (ci), then id
+        let al = a.label.as_deref().unwrap_or("").to_lowercase();
+        let bl = b.label.as_deref().unwrap_or("").to_lowercase();
+        let l = al.cmp(&bl);
+        if l != std::cmp::Ordering::Equal {
+            return l;
+        }
+        a.id.cmp(&b.id)
+    });
+    Ok(rows)
+}

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

@@ -0,0 +1,35 @@
+use codex_tui::ComposerInput;
+
+pub struct NewTaskPage {
+    pub composer: ComposerInput,
+    pub submitting: bool,
+    pub env_id: Option<String>,
+    pub best_of_n: usize,
+}
+
+impl NewTaskPage {
+    pub fn new(env_id: Option<String>, best_of_n: usize) -> Self {
+        let mut composer = ComposerInput::new();
+        composer.set_hint_items(vec![
+            ("⏎", "send"),
+            ("Shift+⏎", "newline"),
+            ("Ctrl+O", "env"),
+            ("Ctrl+N", "attempts"),
+            ("Ctrl+C", "quit"),
+        ]);
+        Self {
+            composer,
+            submitting: false,
+            env_id,
+            best_of_n,
+        }
+    }
+
+    // Additional helpers can be added as usage evolves.
+}
+
+impl Default for NewTaskPage {
+    fn default() -> Self {
+        Self::new(None, 1)
+    }
+}

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

@@ -0,0 +1,176 @@
+use unicode_width::UnicodeWidthChar;
+use unicode_width::UnicodeWidthStr;
+
+/// Scroll position and geometry for a vertical scroll view.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct ScrollViewState {
+    pub scroll: u16,
+    pub viewport_h: u16,
+    pub content_h: u16,
+}
+
+impl ScrollViewState {
+    pub fn clamp(&mut self) {
+        let max_scroll = self.content_h.saturating_sub(self.viewport_h);
+        if self.scroll > max_scroll {
+            self.scroll = max_scroll;
+        }
+    }
+}
+
+/// A simple, local scrollable view for diffs or message text.
+///
+/// Owns raw lines, caches wrapped lines for a given width, and maintains
+/// a small scroll state that is clamped whenever geometry shrinks.
+#[derive(Clone, Debug, Default)]
+pub struct ScrollableDiff {
+    raw: Vec<String>,
+    wrapped: Vec<String>,
+    wrapped_src_idx: Vec<usize>,
+    wrap_cols: Option<u16>,
+    pub state: ScrollViewState,
+}
+
+impl ScrollableDiff {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Replace the raw content lines. Does not rewrap immediately; call `set_width` next.
+    pub fn set_content(&mut self, lines: Vec<String>) {
+        self.raw = lines;
+        self.wrapped.clear();
+        self.wrapped_src_idx.clear();
+        self.state.content_h = 0;
+        // Force rewrap on next set_width even if width is unchanged
+        self.wrap_cols = None;
+    }
+
+    /// Set the wrap width. If changed, rebuild wrapped lines and clamp scroll.
+    pub fn set_width(&mut self, width: u16) {
+        if self.wrap_cols == Some(width) {
+            return;
+        }
+        self.wrap_cols = Some(width);
+        self.rewrap(width);
+        self.state.clamp();
+    }
+
+    /// Update viewport height and clamp scroll if needed.
+    pub fn set_viewport(&mut self, height: u16) {
+        self.state.viewport_h = height;
+        self.state.clamp();
+    }
+
+    /// Return the cached wrapped lines. Call `set_width` first when area changes.
+    pub fn wrapped_lines(&self) -> &[String] {
+        &self.wrapped
+    }
+
+    pub fn wrapped_src_indices(&self) -> &[usize] {
+        &self.wrapped_src_idx
+    }
+
+    pub fn raw_line_at(&self, idx: usize) -> &str {
+        self.raw.get(idx).map(String::as_str).unwrap_or("")
+    }
+
+    /// Scroll by a signed delta; clamps to content.
+    pub fn scroll_by(&mut self, delta: i16) {
+        let s = self.state.scroll as i32 + delta as i32;
+        self.state.scroll = s.clamp(0, self.max_scroll() as i32) as u16;
+    }
+
+    /// Page by a signed delta; typically viewport_h - 1.
+    pub fn page_by(&mut self, delta: i16) {
+        self.scroll_by(delta);
+    }
+
+    pub fn to_top(&mut self) {
+        self.state.scroll = 0;
+    }
+
+    pub fn to_bottom(&mut self) {
+        self.state.scroll = self.max_scroll();
+    }
+
+    /// Optional percent scrolled; None when not enough geometry is known.
+    pub fn percent_scrolled(&self) -> Option<u8> {
+        if self.state.content_h == 0 || self.state.viewport_h == 0 {
+            return None;
+        }
+        if self.state.content_h <= self.state.viewport_h {
+            return None;
+        }
+        let visible_bottom = self.state.scroll.saturating_add(self.state.viewport_h) as f32;
+        let pct = (visible_bottom / self.state.content_h as f32 * 100.0).round();
+        Some(pct.clamp(0.0, 100.0) as u8)
+    }
+
+    fn max_scroll(&self) -> u16 {
+        self.state.content_h.saturating_sub(self.state.viewport_h)
+    }
+
+    fn rewrap(&mut self, width: u16) {
+        if width == 0 {
+            self.wrapped = self.raw.clone();
+            self.state.content_h = self.wrapped.len() as u16;
+            return;
+        }
+        let max_cols = width as usize;
+        let mut out: Vec<String> = Vec::new();
+        let mut out_idx: Vec<usize> = Vec::new();
+        for (raw_idx, raw) in self.raw.iter().enumerate() {
+            // Normalize tabs for width accounting (MVP: 4 spaces).
+            let raw = raw.replace('\t', "    ");
+            if raw.is_empty() {
+                out.push(String::new());
+                out_idx.push(raw_idx);
+                continue;
+            }
+            let mut line = String::new();
+            let mut line_cols = 0usize;
+            let mut last_soft_idx: Option<usize> = None; // last whitespace or punctuation break
+            for (_i, ch) in raw.char_indices() {
+                if ch == '\n' {
+                    out.push(std::mem::take(&mut line));
+                    out_idx.push(raw_idx);
+                    line_cols = 0;
+                    last_soft_idx = None;
+                    continue;
+                }
+                let w = UnicodeWidthChar::width(ch).unwrap_or(0);
+                if line_cols.saturating_add(w) > max_cols {
+                    if let Some(split) = last_soft_idx {
+                        let (prefix, rest) = line.split_at(split);
+                        out.push(prefix.trim_end().to_string());
+                        out_idx.push(raw_idx);
+                        line = rest.trim_start().to_string();
+                        last_soft_idx = None;
+                        // retry add current ch now that line may be shorter
+                    } else if !line.is_empty() {
+                        out.push(std::mem::take(&mut line));
+                        out_idx.push(raw_idx);
+                    }
+                }
+                if ch.is_whitespace()
+                    || matches!(
+                        ch,
+                        ',' | ';' | '.' | ':' | ')' | ']' | '}' | '|' | '/' | '?' | '!' | '-' | '_'
+                    )
+                {
+                    last_soft_idx = Some(line.len());
+                }
+                line.push(ch);
+                line_cols = UnicodeWidthStr::width(line.as_str());
+            }
+            if !line.is_empty() {
+                out.push(line);
+                out_idx.push(raw_idx);
+            }
+        }
+        self.wrapped = out;
+        self.wrapped_src_idx = out_idx;
+        self.state.content_h = self.wrapped.len() as u16;
+    }
+}

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

@@ -0,0 +1,93 @@
+use base64::Engine as _;
+use chrono::Utc;
+use reqwest::header::HeaderMap;
+
+pub fn set_user_agent_suffix(suffix: &str) {
+    if let Ok(mut guard) = codex_core::default_client::USER_AGENT_SUFFIX.lock() {
+        guard.replace(suffix.to_string());
+    }
+}
+
+pub fn append_error_log(message: impl AsRef<str>) {
+    let ts = Utc::now().to_rfc3339();
+    if let Ok(mut f) = std::fs::OpenOptions::new()
+        .create(true)
+        .append(true)
+        .open("error.log")
+    {
+        use std::io::Write as _;
+        let _ = writeln!(f, "[{ts}] {}", message.as_ref());
+    }
+}
+
+/// Normalize the configured base URL to a canonical form used by the backend client.
+/// - trims trailing '/'
+/// - appends '/backend-api' for ChatGPT hosts when missing
+pub fn normalize_base_url(input: &str) -> String {
+    let mut base_url = input.to_string();
+    while base_url.ends_with('/') {
+        base_url.pop();
+    }
+    if (base_url.starts_with("https://chatgpt.com")
+        || base_url.starts_with("https://chat.openai.com"))
+        && !base_url.contains("/backend-api")
+    {
+        base_url = format!("{base_url}/backend-api");
+    }
+    base_url
+}
+
+/// Extract the ChatGPT account id from a JWT token, when present.
+pub fn extract_chatgpt_account_id(token: &str) -> Option<String> {
+    let mut parts = token.split('.');
+    let (_h, payload_b64, _s) = match (parts.next(), parts.next(), parts.next()) {
+        (Some(h), Some(p), Some(s)) if !h.is_empty() && !p.is_empty() && !s.is_empty() => (h, p, s),
+        _ => return None,
+    };
+    let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
+        .decode(payload_b64)
+        .ok()?;
+    let v: serde_json::Value = serde_json::from_slice(&payload_bytes).ok()?;
+    v.get("https://api.openai.com/auth")
+        .and_then(|auth| auth.get("chatgpt_account_id"))
+        .and_then(|id| id.as_str())
+        .map(str::to_string)
+}
+
+/// Build headers for ChatGPT-backed requests: `User-Agent`, optional `Authorization`,
+/// and optional `ChatGPT-Account-Id`.
+pub async fn build_chatgpt_headers() -> HeaderMap {
+    use reqwest::header::AUTHORIZATION;
+    use reqwest::header::HeaderName;
+    use reqwest::header::HeaderValue;
+    use reqwest::header::USER_AGENT;
+
+    set_user_agent_suffix("codex_cloud_tasks_tui");
+    let ua = codex_core::default_client::get_codex_user_agent();
+    let mut headers = HeaderMap::new();
+    headers.insert(
+        USER_AGENT,
+        HeaderValue::from_str(&ua).unwrap_or(HeaderValue::from_static("codex-cli")),
+    );
+    if let Ok(home) = codex_core::config::find_codex_home() {
+        let am = codex_login::AuthManager::new(home);
+        if let Some(auth) = am.auth()
+            && let Ok(tok) = auth.get_token().await
+            && !tok.is_empty()
+        {
+            let v = format!("Bearer {tok}");
+            if let Ok(hv) = HeaderValue::from_str(&v) {
+                headers.insert(AUTHORIZATION, hv);
+            }
+            if let Some(acc) = auth
+                .get_account_id()
+                .or_else(|| extract_chatgpt_account_id(&tok))
+                && let Ok(name) = HeaderName::from_bytes(b"ChatGPT-Account-Id")
+                && let Ok(hv) = HeaderValue::from_str(&acc)
+            {
+                headers.insert(name, hv);
+            }
+        }
+    }
+    headers
+}

codex-rs/cloud-tasks/tests/env_filter.rs

@@ -0,0 +1,22 @@
+use codex_cloud_tasks_client::CloudBackend;
+use codex_cloud_tasks_client::MockClient;
+
+#[tokio::test]
+async fn mock_backend_varies_by_env() {
+    let client = MockClient;
+
+    let root = CloudBackend::list_tasks(&client, None).await.unwrap();
+    assert!(root.iter().any(|t| t.title.contains("Update README")));
+
+    let a = CloudBackend::list_tasks(&client, Some("env-A"))
+        .await
+        .unwrap();
+    assert_eq!(a.len(), 1);
+    assert_eq!(a[0].title, "A: First");
+
+    let b = CloudBackend::list_tasks(&client, Some("env-B"))
+        .await
+        .unwrap();
+    assert_eq!(b.len(), 2);
+    assert!(b[0].title.starts_with("B: "));
+}

codex-rs/codex-backend-openapi-models/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "codex-backend-openapi-models"
+version = { workspace = true }
+edition = "2024"
+
+[lib]
+name = "codex_backend_openapi_models"
+path = "src/lib.rs"
+
+# Important: generated code often violates our workspace lints.
+# Allow unwrap/expect in this crate so the workspace builds cleanly
+# after models are regenerated.
+# Lint overrides are applied in src/lib.rs via crate attributes
+
+[dependencies]
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"

codex-rs/codex-backend-openapi-models/src/lib.rs

@@ -0,0 +1,6 @@
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+// Re-export generated OpenAPI models.
+// The regen script populates `src/models/*.rs` and writes `src/models/mod.rs`.
+// This module intentionally contains no hand-written types.
+pub mod models;

codex-rs/codex-backend-openapi-models/src/models/code_task_details_response.rs

@@ -0,0 +1,42 @@
+/*
+ * codex-backend
+ *
+ * codex-backend
+ *
+ * The version of the OpenAPI document: 0.0.1
+ *
+ * Generated by: https://openapi-generator.tech
+ */
+
+use crate::models;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+pub struct CodeTaskDetailsResponse {
+    #[serde(rename = "task")]
+    pub task: Box<models::TaskResponse>,
+    #[serde(rename = "current_user_turn", skip_serializing_if = "Option::is_none")]
+    pub current_user_turn: Option<std::collections::HashMap<String, serde_json::Value>>,
+    #[serde(
+        rename = "current_assistant_turn",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub current_assistant_turn: Option<std::collections::HashMap<String, serde_json::Value>>,
+    #[serde(
+        rename = "current_diff_task_turn",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub current_diff_task_turn: Option<std::collections::HashMap<String, serde_json::Value>>,
+}
+
+impl CodeTaskDetailsResponse {
+    pub fn new(task: models::TaskResponse) -> CodeTaskDetailsResponse {
+        CodeTaskDetailsResponse {
+            task: Box::new(task),
+            current_user_turn: None,
+            current_assistant_turn: None,
+            current_diff_task_turn: None,
+        }
+    }
+}

codex-rs/codex-backend-openapi-models/src/models/external_pull_request_response.rs

@@ -0,0 +1,40 @@
+/*
+ * codex-backend
+ *
+ * codex-backend
+ *
+ * The version of the OpenAPI document: 0.0.1
+ *
+ * Generated by: https://openapi-generator.tech
+ */
+
+use crate::models;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+pub struct ExternalPullRequestResponse {
+    #[serde(rename = "id")]
+    pub id: String,
+    #[serde(rename = "assistant_turn_id")]
+    pub assistant_turn_id: String,
+    #[serde(rename = "pull_request")]
+    pub pull_request: Box<models::GitPullRequest>,
+    #[serde(rename = "codex_updated_sha", skip_serializing_if = "Option::is_none")]
+    pub codex_updated_sha: Option<String>,
+}
+
+impl ExternalPullRequestResponse {
+    pub fn new(
+        id: String,
+        assistant_turn_id: String,
+        pull_request: models::GitPullRequest,
+    ) -> ExternalPullRequestResponse {
+        ExternalPullRequestResponse {
+            id,
+            assistant_turn_id,
+            pull_request: Box::new(pull_request),
+            codex_updated_sha: None,
+        }
+    }
+}

codex-rs/codex-backend-openapi-models/src/models/git_pull_request.rs

@@ -0,0 +1,77 @@
+/*
+ * codex-backend
+ *
+ * codex-backend
+ *
+ * The version of the OpenAPI document: 0.0.1
+ *
+ * Generated by: https://openapi-generator.tech
+ */
+
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+pub struct GitPullRequest {
+    #[serde(rename = "number")]
+    pub number: i32,
+    #[serde(rename = "url")]
+    pub url: String,
+    #[serde(rename = "state")]
+    pub state: String,
+    #[serde(rename = "merged")]
+    pub merged: bool,
+    #[serde(rename = "mergeable")]
+    pub mergeable: bool,
+    #[serde(rename = "draft", skip_serializing_if = "Option::is_none")]
+    pub draft: Option<bool>,
+    #[serde(rename = "title", skip_serializing_if = "Option::is_none")]
+    pub title: Option<String>,
+    #[serde(rename = "body", skip_serializing_if = "Option::is_none")]
+    pub body: Option<String>,
+    #[serde(rename = "base", skip_serializing_if = "Option::is_none")]
+    pub base: Option<String>,
+    #[serde(rename = "head", skip_serializing_if = "Option::is_none")]
+    pub head: Option<String>,
+    #[serde(rename = "base_sha", skip_serializing_if = "Option::is_none")]
+    pub base_sha: Option<String>,
+    #[serde(rename = "head_sha", skip_serializing_if = "Option::is_none")]
+    pub head_sha: Option<String>,
+    #[serde(rename = "merge_commit_sha", skip_serializing_if = "Option::is_none")]
+    pub merge_commit_sha: Option<String>,
+    #[serde(rename = "comments", skip_serializing_if = "Option::is_none")]
+    pub comments: Option<serde_json::Value>,
+    #[serde(rename = "diff", skip_serializing_if = "Option::is_none")]
+    pub diff: Option<serde_json::Value>,
+    #[serde(rename = "user", skip_serializing_if = "Option::is_none")]
+    pub user: Option<serde_json::Value>,
+}
+
+impl GitPullRequest {
+    pub fn new(
+        number: i32,
+        url: String,
+        state: String,
+        merged: bool,
+        mergeable: bool,
+    ) -> GitPullRequest {
+        GitPullRequest {
+            number,
+            url,
+            state,
+            merged,
+            mergeable,
+            draft: None,
+            title: None,
+            body: None,
+            base: None,
+            head: None,
+            base_sha: None,
+            head_sha: None,
+            merge_commit_sha: None,
+            comments: None,
+            diff: None,
+            user: None,
+        }
+    }
+}

codex-rs/codex-backend-openapi-models/src/models/mod.rs

@@ -0,0 +1,22 @@
+// Curated minimal export list for current workspace usage.
+// NOTE: This file was previously auto-generated by the OpenAPI generator.
+// Currently export only the types referenced by the workspace
+// The process for this will change
+
+pub mod code_task_details_response;
+pub use self::code_task_details_response::CodeTaskDetailsResponse;
+
+pub mod task_response;
+pub use self::task_response::TaskResponse;
+
+pub mod external_pull_request_response;
+pub use self::external_pull_request_response::ExternalPullRequestResponse;
+
+pub mod git_pull_request;
+pub use self::git_pull_request::GitPullRequest;
+
+pub mod task_list_item;
+pub use self::task_list_item::TaskListItem;
+
+pub mod paginated_list_task_list_item_;
+pub use self::paginated_list_task_list_item_::PaginatedListTaskListItem;

codex-rs/codex-backend-openapi-models/src/models/paginated_list_task_list_item_.rs

@@ -0,0 +1,30 @@
+/*
+ * codex-backend
+ *
+ * codex-backend
+ *
+ * The version of the OpenAPI document: 0.0.1
+ *
+ * Generated by: https://openapi-generator.tech
+ */
+
+use crate::models;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+pub struct PaginatedListTaskListItem {
+    #[serde(rename = "items")]
+    pub items: Vec<models::TaskListItem>,
+    #[serde(rename = "cursor", skip_serializing_if = "Option::is_none")]
+    pub cursor: Option<String>,
+}
+
+impl PaginatedListTaskListItem {
+    pub fn new(items: Vec<models::TaskListItem>) -> PaginatedListTaskListItem {
+        PaginatedListTaskListItem {
+            items,
+            cursor: None,
+        }
+    }
+}

codex-rs/codex-backend-openapi-models/src/models/task_list_item.rs

@@ -0,0 +1,63 @@
+/*
+ * codex-backend
+ *
+ * codex-backend
+ *
+ * The version of the OpenAPI document: 0.0.1
+ *
+ * Generated by: https://openapi-generator.tech
+ */
+
+use crate::models;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+pub struct TaskListItem {
+    #[serde(rename = "id")]
+    pub id: String,
+    #[serde(rename = "title")]
+    pub title: String,
+    #[serde(
+        rename = "has_generated_title",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub has_generated_title: Option<bool>,
+    #[serde(rename = "updated_at", skip_serializing_if = "Option::is_none")]
+    pub updated_at: Option<f64>,
+    #[serde(rename = "created_at", skip_serializing_if = "Option::is_none")]
+    pub created_at: Option<f64>,
+    #[serde(
+        rename = "task_status_display",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub task_status_display: Option<std::collections::HashMap<String, serde_json::Value>>,
+    #[serde(rename = "archived")]
+    pub archived: bool,
+    #[serde(rename = "has_unread_turn")]
+    pub has_unread_turn: bool,
+    #[serde(rename = "pull_requests", skip_serializing_if = "Option::is_none")]
+    pub pull_requests: Option<Vec<models::ExternalPullRequestResponse>>,
+}
+
+impl TaskListItem {
+    pub fn new(
+        id: String,
+        title: String,
+        has_generated_title: Option<bool>,
+        archived: bool,
+        has_unread_turn: bool,
+    ) -> TaskListItem {
+        TaskListItem {
+            id,
+            title,
+            has_generated_title,
+            updated_at: None,
+            created_at: None,
+            task_status_display: None,
+            archived,
+            has_unread_turn,
+            pull_requests: None,
+        }
+    }
+}

codex-rs/codex-backend-openapi-models/src/models/task_response.rs

@@ -0,0 +1,62 @@
+/*
+ * codex-backend
+ *
+ * codex-backend
+ *
+ * The version of the OpenAPI document: 0.0.1
+ *
+ * Generated by: https://openapi-generator.tech
+ */
+
+use crate::models;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+pub struct TaskResponse {
+    #[serde(rename = "id")]
+    pub id: String,
+    #[serde(rename = "created_at", skip_serializing_if = "Option::is_none")]
+    pub created_at: Option<f64>,
+    #[serde(rename = "title")]
+    pub title: String,
+    #[serde(
+        rename = "has_generated_title",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub has_generated_title: Option<bool>,
+    #[serde(rename = "current_turn_id", skip_serializing_if = "Option::is_none")]
+    pub current_turn_id: Option<String>,
+    #[serde(rename = "has_unread_turn", skip_serializing_if = "Option::is_none")]
+    pub has_unread_turn: Option<bool>,
+    #[serde(
+        rename = "denormalized_metadata",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub denormalized_metadata: Option<std::collections::HashMap<String, serde_json::Value>>,
+    #[serde(rename = "archived")]
+    pub archived: bool,
+    #[serde(rename = "external_pull_requests")]
+    pub external_pull_requests: Vec<models::ExternalPullRequestResponse>,
+}
+
+impl TaskResponse {
+    pub fn new(
+        id: String,
+        title: String,
+        archived: bool,
+        external_pull_requests: Vec<models::ExternalPullRequestResponse>,
+    ) -> TaskResponse {
+        TaskResponse {
+            id,
+            created_at: None,
+            title,
+            has_generated_title: None,
+            current_turn_id: None,
+            has_unread_turn: None,
+            denormalized_metadata: None,
+            archived,
+            external_pull_requests,
+        }
+    }
+}

codex-rs/common/src/model_presets.rs

@@ -29,7 +29,7 @@ const PRESETS: &[ModelPreset] = &[
         label: "gpt-5-codex medium",
         description: "",
         model: "gpt-5-codex",
-        effort: None,
+        effort: Some(ReasoningEffort::Medium),
     },
     ModelPreset {
         id: "gpt-5-codex-high",

codex-rs/core/Cargo.toml

@@ -15,17 +15,21 @@ workspace = true
 anyhow = { workspace = true }
 askama = { workspace = true }
 async-channel = { workspace = true }
+async-trait = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 codex-apply-patch = { workspace = true }
 codex-file-search = { workspace = true }
 codex-mcp-client = { workspace = true }
+codex-rmcp-client = { workspace = true }
 codex-protocol = { workspace = true }
+codex-otel = { workspace = true, features = ["otel"] }
 dirs = { workspace = true }
 env-flags = { workspace = true }
 eventsource-stream = { workspace = true }
 futures = { workspace = true }
+indexmap = { workspace = true }
 libc = { workspace = true }
 mcp-types = { workspace = true }
 os_info = { workspace = true }
@@ -80,6 +84,7 @@ openssl-sys = { workspace = true, features = ["vendored"] }
 [dev-dependencies]
 assert_cmd = { workspace = true }
 core_test_support = { workspace = true }
+escargot = { workspace = true }
 maplit = { workspace = true }
 predicates = { workspace = true }
 pretty_assertions = { workspace = true }
@@ -87,6 +92,7 @@ tempfile = { workspace = true }
 tokio-test = { workspace = true }
 walkdir = { workspace = true }
 wiremock = { workspace = true }
+tracing-test = {  workspace = true, features = ["no-env-filter"] }
 
 [package.metadata.cargo-shear]
 ignored = ["openssl-sys"]

codex-rs/core/src/apply_patch.rs

@@ -1,13 +1,12 @@
 use crate::codex::Session;
 use crate::codex::TurnContext;
+use crate::function_tool::FunctionCallError;
 use crate::protocol::FileChange;
 use crate::protocol::ReviewDecision;
 use crate::safety::SafetyCheck;
 use crate::safety::assess_patch_safety;
 use codex_apply_patch::ApplyPatchAction;
 use codex_apply_patch::ApplyPatchFileChange;
-use codex_protocol::models::FunctionCallOutputPayload;
-use codex_protocol::models::ResponseInputItem;
 use std::collections::HashMap;
 use std::path::PathBuf;
 
@@ -17,7 +16,7 @@ pub(crate) enum InternalApplyPatchInvocation {
     /// The `apply_patch` call was handled programmatically, without any sort
     /// of sandbox, because the user explicitly approved it. This is the
     /// result to use with the `shell` function call that contained `apply_patch`.
-    Output(ResponseInputItem),
+    Output(Result<String, FunctionCallError>),
 
     /// The `apply_patch` call was approved, either automatically because it
     /// appears that it should be allowed based on the user's sandbox policy
@@ -33,12 +32,6 @@ pub(crate) struct ApplyPatchExec {
     pub(crate) user_explicitly_approved_this_action: bool,
 }
 
-impl From<ResponseInputItem> for InternalApplyPatchInvocation {
-    fn from(item: ResponseInputItem) -> Self {
-        InternalApplyPatchInvocation::Output(item)
-    }
-}
-
 pub(crate) async fn apply_patch(
     sess: &Session,
     turn_context: &TurnContext,
@@ -52,12 +45,13 @@ pub(crate) async fn apply_patch(
         &turn_context.sandbox_policy,
         &turn_context.cwd,
     ) {
-        SafetyCheck::AutoApprove { .. } => {
-            InternalApplyPatchInvocation::DelegateToExec(ApplyPatchExec {
-                action,
-                user_explicitly_approved_this_action: false,
-            })
-        }
+        SafetyCheck::AutoApprove {
+            user_explicitly_approved,
+            ..
+        } => InternalApplyPatchInvocation::DelegateToExec(ApplyPatchExec {
+            action,
+            user_explicitly_approved_this_action: user_explicitly_approved,
+        }),
         SafetyCheck::AskUser => {
             // Compute a readable summary of path changes to include in the
             // approval request so the user can make an informed decision.
@@ -77,25 +71,15 @@ pub(crate) async fn apply_patch(
                     })
                 }
                 ReviewDecision::Denied | ReviewDecision::Abort => {
-                    ResponseInputItem::FunctionCallOutput {
-                        call_id: call_id.to_owned(),
-                        output: FunctionCallOutputPayload {
-                            content: "patch rejected by user".to_string(),
-                            success: Some(false),
-                        },
-                    }
-                    .into()
+                    InternalApplyPatchInvocation::Output(Err(FunctionCallError::RespondToModel(
+                        "patch rejected by user".to_string(),
+                    )))
                 }
             }
         }
-        SafetyCheck::Reject { reason } => ResponseInputItem::FunctionCallOutput {
-            call_id: call_id.to_owned(),
-            output: FunctionCallOutputPayload {
-                content: format!("patch rejected: {reason}"),
-                success: Some(false),
-            },
-        }
-        .into(),
+        SafetyCheck::Reject { reason } => InternalApplyPatchInvocation::Output(Err(
+            FunctionCallError::RespondToModel(format!("patch rejected: {reason}")),
+        )),
     }
 }
 

codex-rs/core/src/auth.rs

@@ -267,6 +267,9 @@ pub fn try_read_auth_json(auth_file: &Path) -> std::io::Result<AuthDotJson> {
 }
 
 pub fn write_auth_json(auth_file: &Path, auth_dot_json: &AuthDotJson) -> std::io::Result<()> {
+    if let Some(parent) = auth_file.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
     let json_data = serde_json::to_string_pretty(auth_dot_json)?;
     let mut options = OpenOptions::new();
     options.truncate(true).write(true).create(true);

codex-rs/core/src/bash.rs

@@ -88,6 +88,21 @@ pub fn try_parse_word_only_commands_sequence(tree: &Tree, src: &str) -> Option<V
     Some(commands)
 }
 
+/// Returns the sequence of plain commands within a `bash -lc "..."` invocation
+/// when the script only contains word-only commands joined by safe operators.
+pub fn parse_bash_lc_plain_commands(command: &[String]) -> Option<Vec<Vec<String>>> {
+    let [bash, flag, script] = command else {
+        return None;
+    };
+
+    if bash != "bash" || flag != "-lc" {
+        return None;
+    }
+
+    let tree = try_parse_bash(script)?;
+    try_parse_word_only_commands_sequence(&tree, script)
+}
+
 fn parse_plain_command_from_node(cmd: tree_sitter::Node, src: &str) -> Option<Vec<String>> {
     if cmd.kind() != "command" {
         return None;

codex-rs/core/src/chat_completions.rs

@@ -1,6 +1,19 @@
 use std::time::Duration;
 
+use crate::ModelProviderInfo;
+use crate::client_common::Prompt;
+use crate::client_common::ResponseEvent;
+use crate::client_common::ResponseStream;
+use crate::error::CodexErr;
+use crate::error::Result;
+use crate::model_family::ModelFamily;
+use crate::openai_tools::create_tools_json_for_chat_completions_api;
+use crate::util::backoff;
 use bytes::Bytes;
+use codex_otel::otel_event_manager::OtelEventManager;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ReasoningItemContent;
+use codex_protocol::models::ResponseItem;
 use eventsource_stream::Eventsource;
 use futures::Stream;
 use futures::StreamExt;
@@ -15,25 +28,13 @@ use tokio::time::timeout;
 use tracing::debug;
 use tracing::trace;
 
-use crate::ModelProviderInfo;
-use crate::client_common::Prompt;
-use crate::client_common::ResponseEvent;
-use crate::client_common::ResponseStream;
-use crate::error::CodexErr;
-use crate::error::Result;
-use crate::model_family::ModelFamily;
-use crate::openai_tools::create_tools_json_for_chat_completions_api;
-use crate::util::backoff;
-use codex_protocol::models::ContentItem;
-use codex_protocol::models::ReasoningItemContent;
-use codex_protocol::models::ResponseItem;
-
 /// Implementation for the classic Chat Completions API.
 pub(crate) async fn stream_chat_completions(
     prompt: &Prompt,
     model_family: &ModelFamily,
     client: &reqwest::Client,
     provider: &ModelProviderInfo,
+    otel_event_manager: &OtelEventManager,
 ) -> Result<ResponseStream> {
     if prompt.output_schema.is_some() {
         return Err(CodexErr::UnsupportedOperation(
@@ -294,10 +295,13 @@ pub(crate) async fn stream_chat_completions(
 
         let req_builder = provider.create_request_builder(client, &None).await?;
 
-        let res = req_builder
-            .header(reqwest::header::ACCEPT, "text/event-stream")
-            .json(&payload)
-            .send()
+        let res = otel_event_manager
+            .log_request(attempt, || {
+                req_builder
+                    .header(reqwest::header::ACCEPT, "text/event-stream")
+                    .json(&payload)
+                    .send()
+            })
             .await;
 
         match res {
@@ -308,6 +312,7 @@ pub(crate) async fn stream_chat_completions(
                     stream,
                     tx_event,
                     provider.stream_idle_timeout(),
+                    otel_event_manager.clone(),
                 ));
                 return Ok(ResponseStream { rx_event });
             }
@@ -351,6 +356,7 @@ async fn process_chat_sse<S>(
     stream: S,
     tx_event: mpsc::Sender<Result<ResponseEvent>>,
     idle_timeout: Duration,
+    otel_event_manager: OtelEventManager,
 ) where
     S: Stream<Item = Result<Bytes>> + Unpin,
 {
@@ -374,7 +380,10 @@ async fn process_chat_sse<S>(
     let mut reasoning_text = String::new();
 
     loop {
-        let sse = match timeout(idle_timeout, stream.next()).await {
+        let sse = match otel_event_manager
+            .log_sse_event(|| timeout(idle_timeout, stream.next()))
+            .await
+        {
             Ok(Some(Ok(ev))) => ev,
             Ok(Some(Err(e))) => {
                 let _ = tx_event

codex-rs/core/src/client.rs

@@ -43,9 +43,11 @@ use crate::model_provider_info::WireApi;
 use crate::openai_model_info::get_model_info;
 use crate::openai_tools::create_tools_json_for_responses_api;
 use crate::protocol::RateLimitSnapshot;
+use crate::protocol::RateLimitWindow;
 use crate::protocol::TokenUsage;
 use crate::token_data::PlanType;
 use crate::util::backoff;
+use codex_otel::otel_event_manager::OtelEventManager;
 use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig;
 use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig;
 use codex_protocol::models::ResponseItem;
@@ -72,6 +74,7 @@ struct Error {
 pub struct ModelClient {
     config: Arc<Config>,
     auth_manager: Option<Arc<AuthManager>>,
+    otel_event_manager: OtelEventManager,
     client: reqwest::Client,
     provider: ModelProviderInfo,
     conversation_id: ConversationId,
@@ -83,6 +86,7 @@ impl ModelClient {
     pub fn new(
         config: Arc<Config>,
         auth_manager: Option<Arc<AuthManager>>,
+        otel_event_manager: OtelEventManager,
         provider: ModelProviderInfo,
         effort: Option<ReasoningEffortConfig>,
         summary: ReasoningSummaryConfig,
@@ -93,6 +97,7 @@ impl ModelClient {
         Self {
             config,
             auth_manager,
+            otel_event_manager,
             client,
             provider,
             conversation_id,
@@ -126,6 +131,7 @@ impl ModelClient {
                     &self.config.model_family,
                     &self.client,
                     &self.provider,
+                    &self.otel_event_manager,
                 )
                 .await?;
 
@@ -162,7 +168,12 @@ impl ModelClient {
         if let Some(path) = &*CODEX_RS_SSE_FIXTURE {
             // short circuit for tests
             warn!(path, "Streaming from fixture");
-            return stream_from_fixture(path, self.provider.clone()).await;
+            return stream_from_fixture(
+                path,
+                self.provider.clone(),
+                self.otel_event_manager.clone(),
+            )
+            .await;
         }
 
         let auth_manager = self.auth_manager.clone();
@@ -183,19 +194,23 @@ impl ModelClient {
 
         let input_with_instructions = prompt.get_formatted_input();
 
-        // Only include `text.verbosity` for GPT-5 family models
-        let text = if self.config.model_family.family == "gpt-5" {
-            create_text_param_for_request(self.config.model_verbosity, &prompt.output_schema)
-        } else {
-            if self.config.model_verbosity.is_some() {
-                warn!(
-                    "model_verbosity is set but ignored for non-gpt-5 model family: {}",
-                    self.config.model_family.family
-                );
+        let verbosity = match &self.config.model_family.family {
+            family if family == "gpt-5" => self.config.model_verbosity,
+            _ => {
+                if self.config.model_verbosity.is_some() {
+                    warn!(
+                        "model_verbosity is set but ignored for non-gpt-5 model family: {}",
+                        self.config.model_family.family
+                    );
+                }
+
+                None
             }
-            None
         };
 
+        // Only include `text.verbosity` for GPT-5 family models
+        let text = create_text_param_for_request(verbosity, &prompt.output_schema);
+
         // In general, we want to explicitly send `store: false` when using the Responses API,
         // but in practice, the Azure Responses API rejects `store: false`:
         //
@@ -224,155 +239,175 @@ impl ModelClient {
         if azure_workaround {
             attach_item_ids(&mut payload_json, &input_with_instructions);
         }
-        let payload_body = serde_json::to_string(&payload_json)?;
 
-        let mut attempt = 0;
-        let max_retries = self.provider.request_max_retries();
-
-        loop {
-            attempt += 1;
-
-            // Always fetch the latest auth in case a prior attempt refreshed the token.
-            let auth = auth_manager.as_ref().and_then(|m| m.auth());
-
-            trace!(
-                "POST to {}: {}",
-                self.provider.get_full_url(&auth),
-                payload_body.as_str()
-            );
-
-            let mut req_builder = self
-                .provider
-                .create_request_builder(&self.client, &auth)
-                .await?;
-
-            req_builder = req_builder
-                .header("OpenAI-Beta", "responses=experimental")
-                // Send session_id for compatibility.
-                .header("conversation_id", self.conversation_id.to_string())
-                .header("session_id", self.conversation_id.to_string())
-                .header(reqwest::header::ACCEPT, "text/event-stream")
-                .json(&payload_json);
-
-            if let Some(auth) = auth.as_ref()
-                && auth.mode == AuthMode::ChatGPT
-                && let Some(account_id) = auth.get_account_id()
+        let max_attempts = self.provider.request_max_retries();
+        for attempt in 0..=max_attempts {
+            match self
+                .attempt_stream_responses(attempt, &payload_json, &auth_manager)
+                .await
             {
-                req_builder = req_builder.header("chatgpt-account-id", account_id);
-            }
-
-            let res = req_builder.send().await;
-            if let Ok(resp) = &res {
-                trace!(
-                    "Response status: {}, cf-ray: {}",
-                    resp.status(),
-                    resp.headers()
-                        .get("cf-ray")
-                        .map(|v| v.to_str().unwrap_or_default())
-                        .unwrap_or_default()
-                );
-            }
-
-            match res {
-                Ok(resp) if resp.status().is_success() => {
-                    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
-
-                    if let Some(snapshot) = parse_rate_limit_snapshot(resp.headers())
-                        && tx_event
-                            .send(Ok(ResponseEvent::RateLimits(snapshot)))
-                            .await
-                            .is_err()
-                    {
-                        debug!("receiver dropped rate limit snapshot event");
-                    }
-
-                    // spawn task to process SSE
-                    let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
-                    tokio::spawn(process_sse(
-                        stream,
-                        tx_event,
-                        self.provider.stream_idle_timeout(),
-                    ));
-
-                    return Ok(ResponseStream { rx_event });
+                Ok(stream) => {
+                    return Ok(stream);
                 }
-                Ok(res) => {
-                    let status = res.status();
-
-                    // Pull out Retry‑After header if present.
-                    let retry_after_secs = res
-                        .headers()
-                        .get(reqwest::header::RETRY_AFTER)
-                        .and_then(|v| v.to_str().ok())
-                        .and_then(|s| s.parse::<u64>().ok());
-
-                    if status == StatusCode::UNAUTHORIZED
-                        && let Some(manager) = auth_manager.as_ref()
-                        && manager.auth().is_some()
-                    {
-                        let _ = manager.refresh_token().await;
+                Err(StreamAttemptError::Fatal(e)) => {
+                    return Err(e);
+                }
+                Err(retryable_attempt_error) => {
+                    if attempt == max_attempts {
+                        return Err(retryable_attempt_error.into_error());
                     }
 
-                    // The OpenAI Responses endpoint returns structured JSON bodies even for 4xx/5xx
-                    // errors. When we bubble early with only the HTTP status the caller sees an opaque
-                    // "unexpected status 400 Bad Request" which makes debugging nearly impossible.
-                    // Instead, read (and include) the response text so higher layers and users see the
-                    // exact error message (e.g. "Unknown parameter: 'input[0].metadata'"). The body is
-                    // small and this branch only runs on error paths so the extra allocation is
-                    // negligible.
-                    if !(status == StatusCode::TOO_MANY_REQUESTS
-                        || status == StatusCode::UNAUTHORIZED
-                        || status.is_server_error())
-                    {
-                        // Surface the error body to callers. Use `unwrap_or_default` per Clippy.
-                        let body = res.text().await.unwrap_or_default();
-                        return Err(CodexErr::UnexpectedStatus(status, body));
-                    }
+                    tokio::time::sleep(retryable_attempt_error.delay(attempt)).await;
+                }
+            }
+        }
 
-                    if status == StatusCode::TOO_MANY_REQUESTS {
-                        let rate_limit_snapshot = parse_rate_limit_snapshot(res.headers());
-                        let body = res.json::<ErrorResponse>().await.ok();
-                        if let Some(ErrorResponse { error }) = body {
-                            if error.r#type.as_deref() == Some("usage_limit_reached") {
-                                // Prefer the plan_type provided in the error message if present
-                                // because it's more up to date than the one encoded in the auth
-                                // token.
-                                let plan_type = error
-                                    .plan_type
-                                    .or_else(|| auth.as_ref().and_then(CodexAuth::get_plan_type));
-                                let resets_in_seconds = error.resets_in_seconds;
-                                return Err(CodexErr::UsageLimitReached(UsageLimitReachedError {
-                                    plan_type,
-                                    resets_in_seconds,
-                                    rate_limits: rate_limit_snapshot,
-                                }));
-                            } else if error.r#type.as_deref() == Some("usage_not_included") {
-                                return Err(CodexErr::UsageNotIncluded);
-                            }
+        unreachable!("stream_responses_attempt should always return");
+    }
+
+    /// Single attempt to start a streaming Responses API call.
+    async fn attempt_stream_responses(
+        &self,
+        attempt: u64,
+        payload_json: &Value,
+        auth_manager: &Option<Arc<AuthManager>>,
+    ) -> std::result::Result<ResponseStream, StreamAttemptError> {
+        // Always fetch the latest auth in case a prior attempt refreshed the token.
+        let auth = auth_manager.as_ref().and_then(|m| m.auth());
+
+        trace!(
+            "POST to {}: {:?}",
+            self.provider.get_full_url(&auth),
+            serde_json::to_string(payload_json)
+        );
+
+        let mut req_builder = self
+            .provider
+            .create_request_builder(&self.client, &auth)
+            .await
+            .map_err(StreamAttemptError::Fatal)?;
+
+        req_builder = req_builder
+            .header("OpenAI-Beta", "responses=experimental")
+            // Send session_id for compatibility.
+            .header("conversation_id", self.conversation_id.to_string())
+            .header("session_id", self.conversation_id.to_string())
+            .header(reqwest::header::ACCEPT, "text/event-stream")
+            .json(payload_json);
+
+        if let Some(auth) = auth.as_ref()
+            && auth.mode == AuthMode::ChatGPT
+            && let Some(account_id) = auth.get_account_id()
+        {
+            req_builder = req_builder.header("chatgpt-account-id", account_id);
+        }
+
+        let res = self
+            .otel_event_manager
+            .log_request(attempt, || req_builder.send())
+            .await;
+
+        if let Ok(resp) = &res {
+            trace!(
+                "Response status: {}, cf-ray: {}",
+                resp.status(),
+                resp.headers()
+                    .get("cf-ray")
+                    .map(|v| v.to_str().unwrap_or_default())
+                    .unwrap_or_default()
+            );
+        }
+
+        match res {
+            Ok(resp) if resp.status().is_success() => {
+                let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
+
+                if let Some(snapshot) = parse_rate_limit_snapshot(resp.headers())
+                    && tx_event
+                        .send(Ok(ResponseEvent::RateLimits(snapshot)))
+                        .await
+                        .is_err()
+                {
+                    debug!("receiver dropped rate limit snapshot event");
+                }
+
+                // spawn task to process SSE
+                let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
+                tokio::spawn(process_sse(
+                    stream,
+                    tx_event,
+                    self.provider.stream_idle_timeout(),
+                    self.otel_event_manager.clone(),
+                ));
+
+                Ok(ResponseStream { rx_event })
+            }
+            Ok(res) => {
+                let status = res.status();
+
+                // Pull out Retry‑After header if present.
+                let retry_after_secs = res
+                    .headers()
+                    .get(reqwest::header::RETRY_AFTER)
+                    .and_then(|v| v.to_str().ok())
+                    .and_then(|s| s.parse::<u64>().ok());
+                let retry_after = retry_after_secs.map(|s| Duration::from_millis(s * 1_000));
+
+                if status == StatusCode::UNAUTHORIZED
+                    && let Some(manager) = auth_manager.as_ref()
+                    && manager.auth().is_some()
+                {
+                    let _ = manager.refresh_token().await;
+                }
+
+                // The OpenAI Responses endpoint returns structured JSON bodies even for 4xx/5xx
+                // errors. When we bubble early with only the HTTP status the caller sees an opaque
+                // "unexpected status 400 Bad Request" which makes debugging nearly impossible.
+                // Instead, read (and include) the response text so higher layers and users see the
+                // exact error message (e.g. "Unknown parameter: 'input[0].metadata'"). The body is
+                // small and this branch only runs on error paths so the extra allocation is
+                // negligible.
+                if !(status == StatusCode::TOO_MANY_REQUESTS
+                    || status == StatusCode::UNAUTHORIZED
+                    || status.is_server_error())
+                {
+                    // Surface the error body to callers. Use `unwrap_or_default` per Clippy.
+                    let body = res.text().await.unwrap_or_default();
+                    return Err(StreamAttemptError::Fatal(CodexErr::UnexpectedStatus(
+                        status, body,
+                    )));
+                }
+
+                if status == StatusCode::TOO_MANY_REQUESTS {
+                    let rate_limit_snapshot = parse_rate_limit_snapshot(res.headers());
+                    let body = res.json::<ErrorResponse>().await.ok();
+                    if let Some(ErrorResponse { error }) = body {
+                        if error.r#type.as_deref() == Some("usage_limit_reached") {
+                            // Prefer the plan_type provided in the error message if present
+                            // because it's more up to date than the one encoded in the auth
+                            // token.
+                            let plan_type = error
+                                .plan_type
+                                .or_else(|| auth.as_ref().and_then(CodexAuth::get_plan_type));
+                            let resets_in_seconds = error.resets_in_seconds;
+                            let codex_err = CodexErr::UsageLimitReached(UsageLimitReachedError {
+                                plan_type,
+                                resets_in_seconds,
+                                rate_limits: rate_limit_snapshot,
+                            });
+                            return Err(StreamAttemptError::Fatal(codex_err));
+                        } else if error.r#type.as_deref() == Some("usage_not_included") {
+                            return Err(StreamAttemptError::Fatal(CodexErr::UsageNotIncluded));
                         }
                     }
-
-                    if attempt > max_retries {
-                        if status == StatusCode::INTERNAL_SERVER_ERROR {
-                            return Err(CodexErr::InternalServerError);
-                        }
-
-                        return Err(CodexErr::RetryLimit(status));
-                    }
-
-                    let delay = retry_after_secs
-                        .map(|s| Duration::from_millis(s * 1_000))
-                        .unwrap_or_else(|| backoff(attempt));
-                    tokio::time::sleep(delay).await;
-                }
-                Err(e) => {
-                    if attempt > max_retries {
-                        return Err(e.into());
-                    }
-                    let delay = backoff(attempt);
-                    tokio::time::sleep(delay).await;
                 }
+
+                Err(StreamAttemptError::RetryableHttpError {
+                    status,
+                    retry_after,
+                })
             }
+            Err(e) => Err(StreamAttemptError::RetryableTransportError(e.into())),
         }
     }
 
@@ -380,6 +415,10 @@ impl ModelClient {
         self.provider.clone()
     }
 
+    pub fn get_otel_event_manager(&self) -> OtelEventManager {
+        self.otel_event_manager.clone()
+    }
+
     /// Returns the currently configured model slug.
     pub fn get_model(&self) -> String {
         self.config.model.clone()
@@ -405,6 +444,47 @@ impl ModelClient {
     }
 }
 
+enum StreamAttemptError {
+    RetryableHttpError {
+        status: StatusCode,
+        retry_after: Option<Duration>,
+    },
+    RetryableTransportError(CodexErr),
+    Fatal(CodexErr),
+}
+
+impl StreamAttemptError {
+    /// attempt is 0-based.
+    fn delay(&self, attempt: u64) -> Duration {
+        // backoff() uses 1-based attempts.
+        let backoff_attempt = attempt + 1;
+        match self {
+            Self::RetryableHttpError { retry_after, .. } => {
+                retry_after.unwrap_or_else(|| backoff(backoff_attempt))
+            }
+            Self::RetryableTransportError { .. } => backoff(backoff_attempt),
+            Self::Fatal(_) => {
+                // Should not be called on Fatal errors.
+                Duration::from_secs(0)
+            }
+        }
+    }
+
+    fn into_error(self) -> CodexErr {
+        match self {
+            Self::RetryableHttpError { status, .. } => {
+                if status == StatusCode::INTERNAL_SERVER_ERROR {
+                    CodexErr::InternalServerError
+                } else {
+                    CodexErr::RetryLimit(status)
+                }
+            }
+            Self::RetryableTransportError(error) => error,
+            Self::Fatal(error) => error,
+        }
+    }
+}
+
 #[derive(Debug, Deserialize, Serialize)]
 struct SseEvent {
     #[serde(rename = "type")]
@@ -414,9 +494,6 @@ struct SseEvent {
     delta: Option<String>,
 }
 
-#[derive(Debug, Deserialize)]
-struct ResponseCreated {}
-
 #[derive(Debug, Deserialize)]
 struct ResponseCompleted {
     id: String,
@@ -488,19 +565,44 @@ fn attach_item_ids(payload_json: &mut Value, original_items: &[ResponseItem]) {
 }
 
 fn parse_rate_limit_snapshot(headers: &HeaderMap) -> Option<RateLimitSnapshot> {
-    let primary_used_percent = parse_header_f64(headers, "x-codex-primary-used-percent")?;
-    let secondary_used_percent = parse_header_f64(headers, "x-codex-secondary-used-percent")?;
-    let primary_to_secondary_ratio_percent =
-        parse_header_f64(headers, "x-codex-primary-over-secondary-limit-percent")?;
-    let primary_window_minutes = parse_header_u64(headers, "x-codex-primary-window-minutes")?;
-    let secondary_window_minutes = parse_header_u64(headers, "x-codex-secondary-window-minutes")?;
+    let primary = parse_rate_limit_window(
+        headers,
+        "x-codex-primary-used-percent",
+        "x-codex-primary-window-minutes",
+        "x-codex-primary-reset-after-seconds",
+    );
 
-    Some(RateLimitSnapshot {
-        primary_used_percent,
-        secondary_used_percent,
-        primary_to_secondary_ratio_percent,
-        primary_window_minutes,
-        secondary_window_minutes,
+    let secondary = parse_rate_limit_window(
+        headers,
+        "x-codex-secondary-used-percent",
+        "x-codex-secondary-window-minutes",
+        "x-codex-secondary-reset-after-seconds",
+    );
+
+    Some(RateLimitSnapshot { primary, secondary })
+}
+
+fn parse_rate_limit_window(
+    headers: &HeaderMap,
+    used_percent_header: &str,
+    window_minutes_header: &str,
+    resets_header: &str,
+) -> Option<RateLimitWindow> {
+    let used_percent: Option<f64> = parse_header_f64(headers, used_percent_header);
+
+    used_percent.and_then(|used_percent| {
+        let window_minutes = parse_header_u64(headers, window_minutes_header);
+        let resets_in_seconds = parse_header_u64(headers, resets_header);
+
+        let has_data = used_percent != 0.0
+            || window_minutes.is_some_and(|minutes| minutes != 0)
+            || resets_in_seconds.is_some_and(|seconds| seconds != 0);
+
+        has_data.then_some(RateLimitWindow {
+            used_percent,
+            window_minutes,
+            resets_in_seconds,
+        })
     })
 }
 
@@ -523,6 +625,7 @@ async fn process_sse<S>(
     stream: S,
     tx_event: mpsc::Sender<Result<ResponseEvent>>,
     idle_timeout: Duration,
+    otel_event_manager: OtelEventManager,
 ) where
     S: Stream<Item = Result<Bytes>> + Unpin,
 {
@@ -534,7 +637,10 @@ async fn process_sse<S>(
     let mut response_error: Option<CodexErr> = None;
 
     loop {
-        let sse = match timeout(idle_timeout, stream.next()).await {
+        let sse = match otel_event_manager
+            .log_sse_event(|| timeout(idle_timeout, stream.next()))
+            .await
+        {
             Ok(Some(Ok(sse))) => sse,
             Ok(Some(Err(e))) => {
                 debug!("SSE Error: {e:#}");
@@ -548,6 +654,21 @@ async fn process_sse<S>(
                         id: response_id,
                         usage,
                     }) => {
+                        if let Some(token_usage) = &usage {
+                            otel_event_manager.sse_event_completed(
+                                token_usage.input_tokens,
+                                token_usage.output_tokens,
+                                token_usage
+                                    .input_tokens_details
+                                    .as_ref()
+                                    .map(|d| d.cached_tokens),
+                                token_usage
+                                    .output_tokens_details
+                                    .as_ref()
+                                    .map(|d| d.reasoning_tokens),
+                                token_usage.total_tokens,
+                            );
+                        }
                         let event = ResponseEvent::Completed {
                             response_id,
                             token_usage: usage.map(Into::into),
@@ -555,12 +676,13 @@ async fn process_sse<S>(
                         let _ = tx_event.send(Ok(event)).await;
                     }
                     None => {
-                        let _ = tx_event
-                            .send(Err(response_error.unwrap_or(CodexErr::Stream(
-                                "stream closed before response.completed".into(),
-                                None,
-                            ))))
-                            .await;
+                        let error = response_error.unwrap_or(CodexErr::Stream(
+                            "stream closed before response.completed".into(),
+                            None,
+                        ));
+                        otel_event_manager.see_event_completed_failed(&error);
+
+                        let _ = tx_event.send(Err(error)).await;
                     }
                 }
                 return;
@@ -664,7 +786,9 @@ async fn process_sse<S>(
                                 response_error = Some(CodexErr::Stream(message, delay));
                             }
                             Err(e) => {
-                                debug!("failed to parse ErrorResponse: {e}");
+                                let error = format!("failed to parse ErrorResponse: {e}");
+                                debug!(error);
+                                response_error = Some(CodexErr::Stream(error, None))
                             }
                         }
                     }
@@ -678,7 +802,9 @@ async fn process_sse<S>(
                             response_completed = Some(r);
                         }
                         Err(e) => {
-                            debug!("failed to parse ResponseCompleted: {e}");
+                            let error = format!("failed to parse ResponseCompleted: {e}");
+                            debug!(error);
+                            response_error = Some(CodexErr::Stream(error, None));
                             continue;
                         }
                     };
@@ -725,6 +851,7 @@ async fn process_sse<S>(
 async fn stream_from_fixture(
     path: impl AsRef<Path>,
     provider: ModelProviderInfo,
+    otel_event_manager: OtelEventManager,
 ) -> Result<ResponseStream> {
     let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
     let f = std::fs::File::open(path.as_ref())?;
@@ -743,6 +870,7 @@ async fn stream_from_fixture(
         stream,
         tx_event,
         provider.stream_idle_timeout(),
+        otel_event_manager,
     ));
     Ok(ResponseStream { rx_event })
 }
@@ -798,6 +926,7 @@ mod tests {
     async fn collect_events(
         chunks: &[&[u8]],
         provider: ModelProviderInfo,
+        otel_event_manager: OtelEventManager,
     ) -> Vec<Result<ResponseEvent>> {
         let mut builder = IoBuilder::new();
         for chunk in chunks {
@@ -807,7 +936,12 @@ mod tests {
         let reader = builder.build();
         let stream = ReaderStream::new(reader).map_err(CodexErr::Io);
         let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(16);
-        tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
+        tokio::spawn(process_sse(
+            stream,
+            tx,
+            provider.stream_idle_timeout(),
+            otel_event_manager,
+        ));
 
         let mut events = Vec::new();
         while let Some(ev) = rx.recv().await {
@@ -821,6 +955,7 @@ mod tests {
     async fn run_sse(
         events: Vec<serde_json::Value>,
         provider: ModelProviderInfo,
+        otel_event_manager: OtelEventManager,
     ) -> Vec<ResponseEvent> {
         let mut body = String::new();
         for e in events {
@@ -837,7 +972,12 @@ mod tests {
 
         let (tx, mut rx) = mpsc::channel::<Result<ResponseEvent>>(8);
         let stream = ReaderStream::new(std::io::Cursor::new(body)).map_err(CodexErr::Io);
-        tokio::spawn(process_sse(stream, tx, provider.stream_idle_timeout()));
+        tokio::spawn(process_sse(
+            stream,
+            tx,
+            provider.stream_idle_timeout(),
+            otel_event_manager,
+        ));
 
         let mut out = Vec::new();
         while let Some(ev) = rx.recv().await {
@@ -846,6 +986,18 @@ mod tests {
         out
     }
 
+    fn otel_event_manager() -> OtelEventManager {
+        OtelEventManager::new(
+            ConversationId::new(),
+            "test",
+            "test",
+            None,
+            Some(AuthMode::ChatGPT),
+            false,
+            "test".to_string(),
+        )
+    }
+
     // ────────────────────────────
     // Tests from `implement-test-for-responses-api-sse-parser`
     // ────────────────────────────
@@ -897,9 +1049,12 @@ mod tests {
             requires_openai_auth: false,
         };
 
+        let otel_event_manager = otel_event_manager();
+
         let events = collect_events(
             &[sse1.as_bytes(), sse2.as_bytes(), sse3.as_bytes()],
             provider,
+            otel_event_manager,
         )
         .await;
 
@@ -957,7 +1112,9 @@ mod tests {
             requires_openai_auth: false,
         };
 
-        let events = collect_events(&[sse1.as_bytes()], provider).await;
+        let otel_event_manager = otel_event_manager();
+
+        let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await;
 
         assert_eq!(events.len(), 2);
 
@@ -991,7 +1148,9 @@ mod tests {
             requires_openai_auth: false,
         };
 
-        let events = collect_events(&[sse1.as_bytes()], provider).await;
+        let otel_event_manager = otel_event_manager();
+
+        let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await;
 
         assert_eq!(events.len(), 1);
 
@@ -1096,7 +1255,9 @@ mod tests {
                 requires_openai_auth: false,
             };
 
-            let out = run_sse(evs, provider).await;
+            let otel_event_manager = otel_event_manager();
+
+            let out = run_sse(evs, provider, otel_event_manager).await;
             assert_eq!(out.len(), case.expected_len, "case {}", case.name);
             assert!(
                 (case.expect_first)(&out[0]),

codex-rs/core/src/codex/compact.rs

@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use super::AgentTask;
 use super::Session;
 use super::TurnContext;
 use super::get_last_assistant_message_from_turn;
@@ -15,7 +14,6 @@ use crate::protocol::Event;
 use crate::protocol::EventMsg;
 use crate::protocol::InputItem;
 use crate::protocol::InputMessageKind;
-use crate::protocol::TaskCompleteEvent;
 use crate::protocol::TaskStartedEvent;
 use crate::protocol::TurnContextItem;
 use crate::truncate::truncate_middle;
@@ -37,17 +35,7 @@ struct HistoryBridgeTemplate<'a> {
     summary_text: &'a str,
 }
 
-pub(super) async fn spawn_compact_task(
-    sess: Arc<Session>,
-    turn_context: Arc<TurnContext>,
-    sub_id: String,
-    input: Vec<InputItem>,
-) {
-    let task = AgentTask::compact(sess.clone(), turn_context, sub_id, input);
-    sess.set_task(task).await;
-}
-
-pub(super) async fn run_inline_auto_compact_task(
+pub(crate) async fn run_inline_auto_compact_task(
     sess: Arc<Session>,
     turn_context: Arc<TurnContext>,
 ) {
@@ -55,15 +43,15 @@ pub(super) async fn run_inline_auto_compact_task(
     let input = vec![InputItem::Text {
         text: SUMMARIZATION_PROMPT.to_string(),
     }];
-    run_compact_task_inner(sess, turn_context, sub_id, input, false).await;
+    run_compact_task_inner(sess, turn_context, sub_id, input).await;
 }
 
-pub(super) async fn run_compact_task(
+pub(crate) async fn run_compact_task(
     sess: Arc<Session>,
     turn_context: Arc<TurnContext>,
     sub_id: String,
     input: Vec<InputItem>,
-) {
+) -> Option<String> {
     let start_event = Event {
         id: sub_id.clone(),
         msg: EventMsg::TaskStarted(TaskStartedEvent {
@@ -71,14 +59,8 @@ pub(super) async fn run_compact_task(
         }),
     };
     sess.send_event(start_event).await;
-    run_compact_task_inner(sess.clone(), turn_context, sub_id.clone(), input, true).await;
-    let event = Event {
-        id: sub_id,
-        msg: EventMsg::TaskComplete(TaskCompleteEvent {
-            last_agent_message: None,
-        }),
-    };
-    sess.send_event(event).await;
+    run_compact_task_inner(sess.clone(), turn_context, sub_id.clone(), input).await;
+    None
 }
 
 async fn run_compact_task_inner(
@@ -86,7 +68,6 @@ async fn run_compact_task_inner(
     turn_context: Arc<TurnContext>,
     sub_id: String,
     input: Vec<InputItem>,
-    remove_task_on_completion: bool,
 ) {
     let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input);
     let turn_input = sess
@@ -112,7 +93,8 @@ async fn run_compact_task_inner(
     sess.persist_rollout_items(&[rollout_item]).await;
 
     loop {
-        let attempt_result = drain_to_completed(&sess, turn_context.as_ref(), &prompt).await;
+        let attempt_result =
+            drain_to_completed(&sess, turn_context.as_ref(), &sub_id, &prompt).await;
 
         match attempt_result {
             Ok(()) => {
@@ -148,21 +130,12 @@ async fn run_compact_task_inner(
         }
     }
 
-    if remove_task_on_completion {
-        sess.remove_task(&sub_id).await;
-    }
-    let history_snapshot = {
-        let state = sess.state.lock().await;
-        state.history.contents()
-    };
+    let history_snapshot = sess.history_snapshot().await;
     let summary_text = get_last_assistant_message_from_turn(&history_snapshot).unwrap_or_default();
     let user_messages = collect_user_messages(&history_snapshot);
     let initial_context = sess.build_initial_context(turn_context.as_ref());
     let new_history = build_compacted_history(initial_context, &user_messages, &summary_text);
-    {
-        let mut state = sess.state.lock().await;
-        state.history.replace(new_history);
-    }
+    sess.replace_history(new_history).await;
 
     let rollout_item = RolloutItem::Compacted(CompactedItem {
         message: summary_text.clone(),
@@ -257,6 +230,7 @@ pub(crate) fn build_compacted_history(
 async fn drain_to_completed(
     sess: &Session,
     turn_context: &TurnContext,
+    sub_id: &str,
     prompt: &Prompt,
 ) -> CodexResult<()> {
     let mut stream = turn_context.client.clone().stream(prompt).await?;
@@ -270,10 +244,14 @@ async fn drain_to_completed(
         };
         match event {
             Ok(ResponseEvent::OutputItemDone(item)) => {
-                let mut state = sess.state.lock().await;
-                state.history.record_items(std::slice::from_ref(&item));
+                sess.record_into_history(std::slice::from_ref(&item)).await;
             }
-            Ok(ResponseEvent::Completed { .. }) => {
+            Ok(ResponseEvent::RateLimits(snapshot)) => {
+                sess.update_rate_limits(sub_id, snapshot).await;
+            }
+            Ok(ResponseEvent::Completed { token_usage, .. }) => {
+                sess.update_token_usage_info(sub_id, turn_context, token_usage.as_ref())
+                    .await;
                 return Ok(());
             }
             Ok(_) => continue,

codex-rs/core/src/command_safety/is_dangerous_command.rs

@@ -0,0 +1,99 @@
+use crate::bash::parse_bash_lc_plain_commands;
+
+pub fn command_might_be_dangerous(command: &[String]) -> bool {
+    if is_dangerous_to_call_with_exec(command) {
+        return true;
+    }
+
+    // Support `bash -lc "<script>"` where the any part of the script might contain a dangerous command.
+    if let Some(all_commands) = parse_bash_lc_plain_commands(command)
+        && all_commands
+            .iter()
+            .any(|cmd| is_dangerous_to_call_with_exec(cmd))
+    {
+        return true;
+    }
+
+    false
+}
+
+fn is_dangerous_to_call_with_exec(command: &[String]) -> bool {
+    let cmd0 = command.first().map(String::as_str);
+
+    match cmd0 {
+        Some(cmd) if cmd.ends_with("git") || cmd.ends_with("/git") => {
+            matches!(command.get(1).map(String::as_str), Some("reset" | "rm"))
+        }
+
+        Some("rm") => matches!(command.get(1).map(String::as_str), Some("-f" | "-rf")),
+
+        // for sudo <cmd> simply do the check for <cmd>
+        Some("sudo") => is_dangerous_to_call_with_exec(&command[1..]),
+
+        // ── anything else ─────────────────────────────────────────────────
+        _ => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn vec_str(items: &[&str]) -> Vec<String> {
+        items.iter().map(std::string::ToString::to_string).collect()
+    }
+
+    #[test]
+    fn git_reset_is_dangerous() {
+        assert!(command_might_be_dangerous(&vec_str(&["git", "reset"])));
+    }
+
+    #[test]
+    fn bash_git_reset_is_dangerous() {
+        assert!(command_might_be_dangerous(&vec_str(&[
+            "bash",
+            "-lc",
+            "git reset --hard"
+        ])));
+    }
+
+    #[test]
+    fn git_status_is_not_dangerous() {
+        assert!(!command_might_be_dangerous(&vec_str(&["git", "status"])));
+    }
+
+    #[test]
+    fn bash_git_status_is_not_dangerous() {
+        assert!(!command_might_be_dangerous(&vec_str(&[
+            "bash",
+            "-lc",
+            "git status"
+        ])));
+    }
+
+    #[test]
+    fn sudo_git_reset_is_dangerous() {
+        assert!(command_might_be_dangerous(&vec_str(&[
+            "sudo", "git", "reset", "--hard"
+        ])));
+    }
+
+    #[test]
+    fn usr_bin_git_is_dangerous() {
+        assert!(command_might_be_dangerous(&vec_str(&[
+            "/usr/bin/git",
+            "reset",
+            "--hard"
+        ])));
+    }
+
+    #[test]
+    fn rm_rf_is_dangerous() {
+        assert!(command_might_be_dangerous(&vec_str(&["rm", "-rf", "/"])));
+    }
+
+    #[test]
+    fn rm_f_is_dangerous() {
+        assert!(command_might_be_dangerous(&vec_str(&["rm", "-f", "/"])));
+    }
+}

codex-rs/core/src/command_safety/is_safe_command.rs

@@ -1,7 +1,14 @@
-use crate::bash::try_parse_bash;
-use crate::bash::try_parse_word_only_commands_sequence;
+use crate::bash::parse_bash_lc_plain_commands;
 
 pub fn is_known_safe_command(command: &[String]) -> bool {
+    #[cfg(target_os = "windows")]
+    {
+        use super::windows_safe_commands::is_safe_command_windows;
+        if is_safe_command_windows(command) {
+            return true;
+        }
+    }
+
     if is_safe_to_call_with_exec(command) {
         return true;
     }
@@ -12,11 +19,7 @@ pub fn is_known_safe_command(command: &[String]) -> bool {
     // introduce side effects ( "&&", "||", ";", and "|" ). If every
     // individual command in the script is itself a known‑safe command, then
     // the composite expression is considered safe.
-    if let [bash, flag, script] = command
-        && bash == "bash"
-        && flag == "-lc"
-        && let Some(tree) = try_parse_bash(script)
-        && let Some(all_commands) = try_parse_word_only_commands_sequence(&tree, script)
+    if let Some(all_commands) = parse_bash_lc_plain_commands(command)
         && !all_commands.is_empty()
         && all_commands
             .iter()
@@ -24,7 +27,6 @@ pub fn is_known_safe_command(command: &[String]) -> bool {
     {
         return true;
     }
-
     false
 }
 

codex-rs/core/src/command_safety/mod.rs

@@ -0,0 +1,4 @@
+pub mod is_dangerous_command;
+pub mod is_safe_command;
+#[cfg(target_os = "windows")]
+pub mod windows_safe_commands;

codex-rs/core/src/command_safety/windows_safe_commands.rs

@@ -0,0 +1,25 @@
+// This is a WIP. This will eventually contain a real list of common safe Windows commands.
+pub fn is_safe_command_windows(_command: &[String]) -> bool {
+    false
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_safe_command_windows;
+
+    fn vec_str(args: &[&str]) -> Vec<String> {
+        args.iter().map(ToString::to_string).collect()
+    }
+
+    #[test]
+    fn everything_is_unsafe() {
+        for cmd in [
+            vec_str(&["powershell.exe", "-NoLogo", "-Command", "echo hello"]),
+            vec_str(&["copy", "foo", "bar"]),
+            vec_str(&["del", "file.txt"]),
+            vec_str(&["powershell.exe", "Get-ChildItem"]),
+        ] {
+            assert!(!is_safe_command_windows(&cmd));
+        }
+    }
+}