feat: spreadsheet part 3 (#13350)

=

.github/workflows/bazel.yml

@@ -28,14 +28,17 @@ jobs:
             target: x86_64-apple-darwin
 
           # Linux
-          - os: ubuntu-24.04-arm
-            target: aarch64-unknown-linux-gnu
           - os: ubuntu-24.04
             target: x86_64-unknown-linux-gnu
-          - os: ubuntu-24.04-arm
-            target: aarch64-unknown-linux-musl
           - os: ubuntu-24.04
             target: x86_64-unknown-linux-musl
+          # 2026-02-27 Bazel tests have been flaky on arm in CI.
+          # Disable until we can investigate and stabilize them.
+          # - os: ubuntu-24.04-arm
+          #   target: aarch64-unknown-linux-musl
+          # - os: ubuntu-24.04-arm
+          #   target: aarch64-unknown-linux-gnu
+
           # TODO: Enable Windows once we fix the toolchain issues there.
           #- os: windows-latest
           #  target: x86_64-pc-windows-gnullvm

.github/workflows/issue-deduplicator.yml

@@ -7,15 +7,17 @@ on:
       - labeled
 
 jobs:
-  gather-duplicates:
-    name: Identify potential duplicates
+  gather-duplicates-all:
+    name: Identify potential duplicates (all issues)
     # Prevent runs on forks (requires OpenAI API key, wastes Actions minutes)
     if: github.repository == 'openai/codex' && (github.event.action == 'opened' || (github.event.action == 'labeled' && github.event.label.name == 'codex-deduplicate'))
     runs-on: ubuntu-latest
     permissions:
       contents: read
     outputs:
-      codex_output: ${{ steps.select-final.outputs.codex_output }}
+      issues_json: ${{ steps.normalize-all.outputs.issues_json }}
+      reason: ${{ steps.normalize-all.outputs.reason }}
+      has_matches: ${{ steps.normalize-all.outputs.has_matches }}
     steps:
       - uses: actions/checkout@v6
 
@@ -29,7 +31,6 @@ jobs:
 
           CURRENT_ISSUE_FILE=codex-current-issue.json
           EXISTING_ALL_FILE=codex-existing-issues-all.json
-          EXISTING_OPEN_FILE=codex-existing-issues-open.json
 
           gh issue list --repo "$REPO" \
             --json number,title,body,createdAt,updatedAt,state,labels \
@@ -47,22 +48,6 @@ jobs:
               }]' \
             > "$EXISTING_ALL_FILE"
 
-          gh issue list --repo "$REPO" \
-            --json number,title,body,createdAt,updatedAt,state,labels \
-            --limit 1000 \
-            --state open \
-            --search "sort:created-desc" \
-            | jq '[.[] | {
-                number,
-                title,
-                body: ((.body // "")[0:4000]),
-                createdAt,
-                updatedAt,
-                state,
-                labels: ((.labels // []) | map(.name))
-              }]' \
-            > "$EXISTING_OPEN_FILE"
-
           gh issue view "$ISSUE_NUMBER" \
             --repo "$REPO" \
             --json number,title,body \
@@ -71,7 +56,6 @@ jobs:
 
           echo "Prepared duplicate detection input files."
           echo "all_issue_count=$(jq 'length' "$EXISTING_ALL_FILE")"
-          echo "open_issue_count=$(jq 'length' "$EXISTING_OPEN_FILE")"
 
       # Prompt instructions are intentionally inline in this workflow. The old
       # .github/prompts/issue-deduplicator.txt file is obsolete and removed.
@@ -158,9 +142,59 @@ jobs:
             echo "has_matches=$has_matches"
           } >> "$GITHUB_OUTPUT"
 
+  gather-duplicates-open:
+    name: Identify potential duplicates (open issues fallback)
+    # Pass 1 may drop sudo on the runner, so run the fallback in a fresh job.
+    needs: gather-duplicates-all
+    if: ${{ needs.gather-duplicates-all.result == 'success' && needs.gather-duplicates-all.outputs.has_matches != 'true' }}
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    outputs:
+      issues_json: ${{ steps.normalize-open.outputs.issues_json }}
+      reason: ${{ steps.normalize-open.outputs.reason }}
+      has_matches: ${{ steps.normalize-open.outputs.has_matches }}
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Prepare Codex inputs
+        env:
+          GH_TOKEN: ${{ github.token }}
+          REPO: ${{ github.repository }}
+          ISSUE_NUMBER: ${{ github.event.issue.number }}
+        run: |
+          set -eo pipefail
+
+          CURRENT_ISSUE_FILE=codex-current-issue.json
+          EXISTING_OPEN_FILE=codex-existing-issues-open.json
+
+          gh issue list --repo "$REPO" \
+            --json number,title,body,createdAt,updatedAt,state,labels \
+            --limit 1000 \
+            --state open \
+            --search "sort:created-desc" \
+            | jq '[.[] | {
+                number,
+                title,
+                body: ((.body // "")[0:4000]),
+                createdAt,
+                updatedAt,
+                state,
+                labels: ((.labels // []) | map(.name))
+              }]' \
+            > "$EXISTING_OPEN_FILE"
+
+          gh issue view "$ISSUE_NUMBER" \
+            --repo "$REPO" \
+            --json number,title,body \
+            | jq '{number, title, body: ((.body // "")[0:4000])}' \
+            > "$CURRENT_ISSUE_FILE"
+
+          echo "Prepared fallback duplicate detection input files."
+          echo "open_issue_count=$(jq 'length' "$EXISTING_OPEN_FILE")"
+
       - id: codex-open
         name: Find duplicates (pass 2, open issues)
-        if: ${{ steps.normalize-all.outputs.has_matches != 'true' }}
         uses: openai/codex-action@main
         with:
           openai-api-key: ${{ secrets.CODEX_OPENAI_API_KEY }}
@@ -200,7 +234,6 @@ jobs:
 
       - id: normalize-open
         name: Normalize pass 2 output
-        if: ${{ steps.normalize-all.outputs.has_matches != 'true' }}
         env:
           CODEX_OUTPUT: ${{ steps.codex-open.outputs.final-message }}
           CURRENT_ISSUE_NUMBER: ${{ github.event.issue.number }}
@@ -243,15 +276,27 @@ jobs:
             echo "has_matches=$has_matches"
           } >> "$GITHUB_OUTPUT"
 
+  select-final:
+    name: Select final duplicate set
+    needs:
+      - gather-duplicates-all
+      - gather-duplicates-open
+    if: ${{ always() && needs.gather-duplicates-all.result == 'success' && (needs.gather-duplicates-open.result == 'success' || needs.gather-duplicates-open.result == 'skipped') }}
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    outputs:
+      codex_output: ${{ steps.select-final.outputs.codex_output }}
+    steps:
       - id: select-final
         name: Select final duplicate set
         env:
-          PASS1_ISSUES: ${{ steps.normalize-all.outputs.issues_json }}
-          PASS1_REASON: ${{ steps.normalize-all.outputs.reason }}
-          PASS2_ISSUES: ${{ steps.normalize-open.outputs.issues_json }}
-          PASS2_REASON: ${{ steps.normalize-open.outputs.reason }}
-          PASS1_HAS_MATCHES: ${{ steps.normalize-all.outputs.has_matches }}
-          PASS2_HAS_MATCHES: ${{ steps.normalize-open.outputs.has_matches }}
+          PASS1_ISSUES: ${{ needs.gather-duplicates-all.outputs.issues_json }}
+          PASS1_REASON: ${{ needs.gather-duplicates-all.outputs.reason }}
+          PASS2_ISSUES: ${{ needs.gather-duplicates-open.outputs.issues_json }}
+          PASS2_REASON: ${{ needs.gather-duplicates-open.outputs.reason }}
+          PASS1_HAS_MATCHES: ${{ needs.gather-duplicates-all.outputs.has_matches }}
+          PASS2_HAS_MATCHES: ${{ needs.gather-duplicates-open.outputs.has_matches }}
         run: |
           set -eo pipefail
 
@@ -289,8 +334,8 @@ jobs:
 
   comment-on-issue:
     name: Comment with potential duplicates
-    needs: gather-duplicates
-    if: ${{ needs.gather-duplicates.result != 'skipped' }}
+    needs: select-final
+    if: ${{ needs.select-final.result != 'skipped' }}
     runs-on: ubuntu-latest
     permissions:
       contents: read
@@ -299,7 +344,7 @@ jobs:
       - name: Comment on issue
         uses: actions/github-script@v8
         env:
-          CODEX_OUTPUT: ${{ needs.gather-duplicates.outputs.codex_output }}
+          CODEX_OUTPUT: ${{ needs.select-final.outputs.codex_output }}
         with:
           github-token: ${{ github.token }}
           script: |

.github/workflows/shell-tool-mcp.yml

@@ -146,9 +146,8 @@ jobs:
         shell: bash
         run: |
           set -euo pipefail
-          git clone --depth 1 https://github.com/bolinfest/bash /tmp/bash
+          git clone https://git.savannah.gnu.org/git/bash /tmp/bash
           cd /tmp/bash
-          git fetch --depth 1 origin a8a1c2fac029404d3f42cd39f5a20f24b6e4fe4b
           git checkout a8a1c2fac029404d3f42cd39f5a20f24b6e4fe4b
           git apply "${GITHUB_WORKSPACE}/shell-tool-mcp/patches/bash-exec-wrapper.patch"
           ./configure --without-bash-malloc
@@ -188,9 +187,8 @@ jobs:
         shell: bash
         run: |
           set -euo pipefail
-          git clone --depth 1 https://github.com/bolinfest/bash /tmp/bash
+          git clone https://git.savannah.gnu.org/git/bash /tmp/bash
           cd /tmp/bash
-          git fetch --depth 1 origin a8a1c2fac029404d3f42cd39f5a20f24b6e4fe4b
           git checkout a8a1c2fac029404d3f42cd39f5a20f24b6e4fe4b
           git apply "${GITHUB_WORKSPACE}/shell-tool-mcp/patches/bash-exec-wrapper.patch"
           ./configure --without-bash-malloc

app_server_tracing_design.md

@@ -1,691 +0,0 @@
-# App-server v2 tracing design
-
-This document proposes a simple, staged tracing design for
-`codex-rs/app-server` with these goals:
-
-- support distributed tracing from client-initiated app-server work into
-  app-server and `codex-core`
-- keep tracing consistent across the app-server v2 surface area
-- minimize tracing boilerplate in request handlers
-- avoid introducing tracing-owned lifecycle state that duplicates existing
-  app-server runtime state
-
-This design explicitly avoids a `RequestKind` taxonomy and avoids
-app-server-owned long-lived lifecycle span registries.
-
-## Summary
-
-The design has four major pieces:
-
-1. A transport-level W3C trace carrier on inbound JSON-RPC request envelopes.
-2. A centralized app-server request tracing layer that wraps every inbound
-   request in the same request span.
-3. An internal trace-context handoff through `codex_protocol::Submission` so
-   work that continues in `codex-core` inherits the inbound app-server request
-   ancestry.
-4. A core-owned long-lived turn span for turn-producing operations such as
-   `turn/start` and `review/start`.
-
-Every inbound JSON-RPC request gets a standardized request span.
-
-When an app-server request submits work into core, the current span context is
-captured into `Submission.trace`. Core then creates a short-lived dispatch span
-parented from that carrier and, for turn-producing operations, creates a
-long-lived turn span beneath it before continuing into its existing task and
-model request tracing.
-
-Important:
-
-- request spans stay short-lived
-- long-lived turn spans are owned by core, not app-server
-- the design does not add app-server-owned long-lived thread or realtime spans
-
-## Design goals
-
-- **Distributed tracing first**
-  - Clients should be able to send trace context to app-server.
-  - App-server should preserve that trace ancestry across the async handoff into
-    core.
-  - Existing core model request tracing should continue to inherit from the
-    active core span once the handoff occurs.
-
-- **Consistent request instrumentation**
-  - Every inbound request should produce the same request span with the same
-    base attributes.
-  - Request tracing should be wired at the transport boundary, not repeated in
-    individual handlers.
-
-- **Minimal boilerplate**
-  - Request handlers should not manually parse carriers or build request spans.
-  - Existing calls to `thread.submit(...)` and similar APIs should pick up trace
-    propagation automatically.
-
-- **Minimal business logic pollution**
-  - W3C parsing, OTEL conversion, and span-parenting rules should live in
-    tracing-specific modules.
-  - App-server business logic should stay focused on request handling, not span
-    management.
-
-- **Incremental rollout**
-  - The first rollout should prove inbound request tracing and app-server ->
-    core propagation.
-  - Once propagation is in place, core should add a long-lived turn span so a
-    single span covers the actual duration of a turn.
-  - Thread and realtime lifecycle tracing should wait until there is a concrete
-    need.
-
-## Non-goals
-
-- This design does not attempt to make every loaded thread or realtime session
-  correspond to a long-lived tracing span.
-- This design does not add tracing-owned thread or realtime state stores in the
-  initial design.
-- This design does not require every app-server v2 `*Params` type to carry
-  trace metadata.
-- This design does not require outbound JSON-RPC trace propagation in the
-  initial rollout.
-
-## Why not `RequestKind`
-
-An earlier direction considered a central `RequestKind` taxonomy such as
-`Unary`, `TurnLifecycle`, or `RealtimeLifecycle`.
-
-That is workable, but it makes tracing depend on a classification that can
-drift from runtime behavior. The simpler design instead treats tracing as two
-generic mechanics:
-
-- every inbound request gets the same request span
-- any async work that crosses from app-server into core gets the current span
-  context attached to `Submission`
-
-This keeps the initial implementation small and avoids turning tracing into a
-taxonomy maintenance problem.
-
-## Terminology
-
-- **Request span**
-  - A short-lived span for one inbound JSON-RPC request to app-server.
-
-- **W3C trace context**
-  - A serializable representation of distributed trace context based on
-    `traceparent` and `tracestate`.
-
-- **Submission trace handoff**
-  - The optional serialized trace context attached to
-    `codex_protocol::Submission` so core can restore parentage after the
-    app-server request handler returns.
-
-- **Dispatch span**
-  - A short-lived core span created when the submission loop receives a
-    `Submission` with trace context.
-
-- **Turn span**
-  - A long-lived core-owned span representing the actual runtime of a turn from
-    turn start until completion, interruption, or failure.
-
-## High-level tracing model
-
-### 1. Inbound request
-
-For every inbound JSON-RPC request:
-
-1. parse an optional W3C trace carrier from the JSON-RPC envelope
-2. create a standardized request span
-3. parent that span from the incoming carrier when present
-4. process the request inside that span
-
-This is true for every request, regardless of whether the API is unary or
-starts work that continues later.
-
-### 2. Async handoff into core
-
-Some app-server requests submit work that continues in core after the original
-request returns. The critical example is `turn/start`, but the mechanism should
-be generic.
-
-To preserve trace ancestry:
-
-- add an optional `W3cTraceContext` to `codex_protocol::Submission`
-- have `CodexThread::submit()` capture the current span context into that field
-  automatically
-- have `codex-core` create a per-submission dispatch span parented from that
-  carrier
-
-This gives a clean causal chain:
-
-- client span
-- app-server request span
-- core dispatch span
-- core turn span for turn-producing operations
-- existing core spans such as `run_turn`, sampling, and model request spans
-
-### 3. Core-owned turn spans
-
-For turn-producing operations such as `turn/start` and `review/start`:
-
-- app-server creates the inbound request span
-- app-server propagates that request context through `Submission.trace`
-- core creates a dispatch span when it receives the submission
-- core then creates a long-lived turn span beneath that dispatch span
-- existing core work such as `run_turn` and model request tracing runs beneath
-  the turn span
-
-This keeps long-lived span ownership with the layer that actually owns turn
-execution and completion.
-
-### 4. Defer thread and realtime lifecycle-heavy tracing
-
-The design should not add:
-
-- app-server-owned thread residency stores
-- app-server-owned realtime session stores
-
-App-server already maintains thread subscription and runtime state in existing
-structures. If later tracing work needs thread loaded-duration or realtime
-duration metrics, that data should extend those existing structures rather than
-introducing a parallel tracing-only state machine.
-
-## Span model by API shape
-
-The initial implementation keeps the app-server side uniform.
-
-### Unary request/response APIs
-
-Examples:
-
-- `thread/list`
-- `thread/read`
-- `model/list`
-- `config/read`
-- `skills/list`
-- `app/list`
-
-Behavior:
-
-- create request span
-- return response
-- no additional app-server span state
-
-### Turn-producing APIs
-
-Examples:
-
-- `turn/start`
-- `review/start`
-- `thread/compact/start` when it executes as a normal turn lifecycle
-
-Behavior:
-
-- create request span
-- submit work under that request span
-- capture the current span context into `Submission.trace`
-- let core create a dispatch span and then a long-lived turn span
-- let the turn span remain open until the real core turn lifecycle ends
-
-Important: request spans should not stay open until eventual streamed
-completion. The request span ends quickly; the core-owned turn span carries the
-long-running work.
-
-### Other APIs that submit work into core
-
-Examples:
-
-- `thread/realtime/start`
-- `thread/realtime/appendAudio`
-- `thread/realtime/appendText`
-- `thread/realtime/stop`
-
-Behavior:
-
-- create request span
-- submit work under that request span
-- capture the current span context into `Submission.trace`
-- let core continue tracing from there
-
-These APIs do not automatically imply a long-lived app-server or core lifecycle
-span in the initial design.
-
-### Thread lifecycle APIs
-
-Examples:
-
-- `thread/start`
-- `thread/resume`
-- `thread/fork`
-- `thread/unsubscribe`
-
-Behavior in the initial design:
-
-- create request span
-- annotate with `thread.id` when known
-- do not introduce separate app-server lifecycle spans or tracing-only state
-
-If later work needs thread loaded/unloaded metrics, it should reuse the existing
-thread runtime state already maintained by app-server.
-
-## Where the code should live
-
-### `codex-rs/protocol`
-
-Add a small shared `W3cTraceContext` type to
-[`codex-rs/protocol/src/protocol.rs`](/Users/owen/repos/codex3/codex-rs/protocol/src/protocol.rs).
-
-Responsibilities:
-
-- define a serializable W3C trace context type
-- avoid direct dependence on OTEL runtime types
-- be usable from both protocol crates and runtime crates
-
-Suggested contents:
-
-- `W3cTraceContext`
-  - `traceparent: Option<String>`
-  - `tracestate: Option<String>`
-
-Suggested `Submission` change:
-
-- `Submission { id, op, trace: Option<W3cTraceContext> }`
-
-This is the only new internal async handoff needed for the initial rollout.
-
-### `codex-rs/otel`
-
-Add a small helper module or extend existing tracing helpers so OTEL-specific
-logic stays centralized.
-
-Responsibilities:
-
-- convert `W3cTraceContext` -> OTEL `Context`
-- convert the current tracing span context -> `W3cTraceContext`
-- parent a tracing span from an explicit carrier when present
-- apply precedence rules:
-  - explicit carrier from app-server transport or `Submission.trace`
-  - fallback to env `TRACEPARENT` / `TRACESTATE`
-  - otherwise root span
-
-Important:
-
-- keep this focused on carrier parsing and span parenting
-- do not move app-server runtime state into `codex-otel`
-- do not overload `OtelManager` with app-server lifecycle ownership in the
-  initial design
-
-### `codex-rs/app-server-protocol`
-
-Extend inbound JSON-RPC request envelopes in
-[`codex-rs/app-server-protocol/src/jsonrpc_lite.rs`](/Users/owen/repos/codex3/codex-rs/app-server-protocol/src/jsonrpc_lite.rs)
-with a dedicated optional trace carrier field.
-
-Suggested shape:
-
-- `JSONRPCRequest { id, method, params, trace }`
-
-Where:
-
-- `trace: Option<W3cTraceContext>`
-
-Important:
-
-- use a dedicated tracing field, not a generic `meta` bag
-- keep tracing transport-level and method-agnostic
-- do not add trace fields to individual `*Params` business payloads
-
-### `codex-rs/core`
-
-Make small changes in the submission path in
-[`codex-rs/core/src/codex.rs`](/Users/owen/repos/codex3/codex-rs/core/src/codex.rs).
-
-Responsibilities:
-
-- read `Submission.trace`
-- create a per-submission dispatch span parented from that carrier
-- run existing submission handling under that span
-
-This is enough for existing core tracing to inherit the correct ancestry, and
-it is the right place to add the long-lived turn span required for turn
-lifecycles.
-
-For turn-producing operations, core responsibilities should include:
-
-- read `Submission.trace`
-- create a per-submission dispatch span parented from that carrier
-- create a long-lived turn span beneath the dispatch span when the operation
-  actually starts a turn
-- finish that turn span when the real core turn lifecycle completes,
-  interrupts, or fails
-
-### `codex-rs/app-server`
-
-Add a small dedicated tracing module rather than spreading request tracing logic
-across handlers. A likely shape is:
-
-- `app_server_tracing/mod.rs`
-- `app_server_tracing/request_spans.rs`
-- `app_server_tracing/incoming.rs`
-
-Responsibilities:
-
-- extract incoming W3C trace carriers from JSON-RPC requests
-- build standardized request spans
-- provide a small API that wraps request handling in the correct span
-
-Non-responsibilities in the initial design:
-
-- no thread residency registry
-- no realtime session registry
-
-## Standardized request spans
-
-Every inbound request should use the same request-span builder.
-
-Suggested name:
-
-- `app_server.request`
-
-Suggested attributes:
-
-- `rpc.system = "jsonrpc"`
-- `rpc.service = "codex-app-server"`
-- `rpc.method`
-- `rpc.transport`
-  - `stdio`
-  - `websocket`
-- `rpc.request_id`
-- `app_server.connection_id`
-- `app_server.api_version = "v2"` when applicable
-- `app_server.client_name` when known from initialize
-- `app_server.client_version` when known
-
-Optional useful attributes:
-
-- `thread.id` when already known from params
-- `turn.id` when already known from params
-
-Important:
-
-- the span factory should be the only place that assembles these fields
-- handlers should not manually construct request-span attributes
-- for the `initialize` request itself, read `clientInfo.name` and
-  `clientInfo.version` directly from the request params when present
-- for later requests on the same connection, read client metadata from
-  per-connection session state populated during `initialize`
-
-## No app-server tracing registries
-
-The design should not introduce app-server-owned tracing registries for turns,
-threads, or realtime sessions.
-
-Why:
-
-- app-server already has thread subscription and runtime state
-- core already owns the real task and turn lifecycle
-- a second tracing-specific state machine adds more code and more ways for
-  lifecycle tracking to drift
-
-Future guidance:
-
-- if thread loaded/unloaded metrics become important, extend existing app-server
-  thread state
-- keep long-lived turn spans in core
-- if realtime lifecycle metrics become important, extend the existing realtime
-  runtime path rather than creating a parallel tracing store
-
-## No direct span construction in handlers
-
-Request handlers should not call `info_span!`, `trace_span!`, `set_parent`, or
-OTEL APIs directly for app-server request tracing.
-
-Instead:
-
-- `message_processor` should wrap inbound request handling through the
-  centralized request-span helper
-- `CodexThread::submit()` should capture the current span context into
-  `Submission.trace`
-
-That keeps request tracing transport-level and largely invisible to business
-handlers.
-
-## Layering
-
-The intended call graph is:
-
-- `message_processor` -> `app_server_tracing`
-  - create and enter the standardized inbound request span
-- `CodexThread::submit()` -> `codex-otel` trace-context helper
-  - snapshot the current span context into `Submission.trace`
-- `codex-core` submission loop -> `codex-otel` trace-context helper
-  - create a dispatch span parented from `Submission.trace`
-  - create a long-lived turn span for turn-producing operations
-
-Important:
-
-- app-server owns inbound request tracing
-- core owns execution after the async handoff
-- core owns long-lived turn spans
-- the design does not add app-server-owned long-lived thread or realtime spans
-
-## Inbound flow in app-server
-
-The inbound request path should work like this:
-
-1. Parse the JSON-RPC request envelope, including `trace`.
-2. Use the tracing module to create a request span.
-3. Process the request inside that span.
-4. If the request submits work into core, let `CodexThread::submit()` capture
-   the active span context into `Submission.trace`.
-
-Integration point:
-
-- [`codex-rs/app-server/src/message_processor.rs`](/Users/owen/repos/codex3/codex-rs/app-server/src/message_processor.rs)
-
-## Core handoff flow
-
-The `turn/start` and similar flows cross an async boundary:
-
-- app-server handler submits work
-- core submission loop receives `Submission`
-- actual work continues later on different tasks
-
-To preserve parentage:
-
-1. app-server request handling runs inside `app_server.request`
-2. `CodexThread::submit()` captures that active context into `Submission.trace`
-3. core submission loop creates a dispatch span parented from `Submission.trace`
-4. if the submission starts a turn, core creates a long-lived turn span beneath
-   that dispatch span
-5. existing core spans naturally nest under the turn span
-
-This lets:
-
-- submission handling
-- a single long-lived turn span for turn-producing APIs
-- `run_turn`
-- model client request tracing
-
-inherit the app-server request trace without broad tracing changes across core.
-
-## Behavior for key v2 APIs
-
-### `thread/start`
-
-- create request span
-- annotate with `thread.id` once known
-- send response and `thread/started`
-- no separate thread lifecycle span in the initial design
-
-### `thread/resume`
-
-- create request span
-- annotate with `thread.id` when known
-- no separate lifecycle span
-
-### `thread/fork`
-
-- create request span
-- annotate with the new `thread.id`
-- no separate lifecycle span
-
-### `thread/unsubscribe`
-
-- create request span
-- no separate unload span
-- if later thread unload metrics are needed, reuse existing thread state rather
-  than adding a tracing-only registry
-
-### `turn/start`
-
-- create request span
-- submit work into core under that request span
-- propagate the active span context through `Submission.trace`
-- let core create a dispatch span and then a long-lived turn span
-- let that turn span cover the full duration until completion, interruption, or
-  failure
-
-### `turn/steer`
-
-- create request span
-- if the request submits core work, propagate via `Submission.trace`
-- otherwise request span only
-
-### `turn/interrupt`
-
-- create request span
-- request span only unless core submission is involved
-
-### `review/start`
-
-- treat like `turn/start`
-- let core create the same kind of long-lived turn span
-
-### `thread/realtime/start`, `appendAudio`, `appendText`, `stop`
-
-- create request span
-- if the API submits work into core, propagate via `Submission.trace`
-- do not introduce separate realtime lifecycle spans in the initial design
-
-### Unary methods such as `thread/list`
-
-- create request span only
-
-## Runtime checks
-
-Keep runtime checks narrowly scoped in the initial rollout:
-
-- warn when an inbound trace carrier is present but invalid
-- test that `Submission.trace` is set when work is submitted from a traced
-  request
-
-Do not add lifecycle consistency checks for tracing registries that do not
-exist yet.
-
-## Tests
-
-Add tests for the initial mechanics:
-
-- inbound request tracing accepts a valid W3C carrier
-- invalid carriers are ignored cleanly
-- unary methods create request spans without needing any extra handler changes
-- `turn/start` propagates request ancestry through `Submission.trace` into core
-- `turn/start` creates a long-lived core-owned turn span
-- the turn span closes on completion, interruption, or failure
-- existing core spans inherit from the propagated parent
-
-The goal is to verify the centralized propagation behavior, not to exhaustively
-test OTEL internals.
-
-## Suggested PR sequence
-
-### PR 1: Foundation plus inbound request spans
-
-Scope:
-
-1. Introduce a shared `W3cTraceContext` type in `codex-protocol`.
-2. Add `trace` to inbound JSON-RPC request envelopes in app-server protocol.
-3. Add focused trace-context helpers in `codex-rs/otel`.
-4. Add the centralized app-server request tracing module.
-5. Wrap inbound request handling in `message_processor.rs`.
-
-Why this PR:
-
-- proves the transport and request-span shape with minimal scope
-- gives all inbound app-server APIs consistent request tracing immediately
-- avoids mixing lifecycle questions into the initial plumbing review
-
-### PR 2: Async handoff into core via `Submission`
-
-Scope:
-
-1. Add `trace` to `Submission`.
-2. Have `CodexThread::submit()` capture the current span context automatically.
-3. Have the core submission loop restore parentage with a dispatch span.
-4. Validate the flow with `turn/start`.
-
-Why this PR:
-
-- validates the critical async handoff from app-server into core
-- proves that existing core tracing can inherit the app-server request ancestry
-- keeps the behavior change focused on one boundary
-
-### PR 3: Core-owned long-lived turn spans
-
-Scope:
-
-1. Add a long-lived turn span in core for `turn/start`.
-2. Reuse the same turn-span pattern for `review/start`.
-3. Ensure the span closes on completion, interruption, or failure.
-
-Why this PR:
-
-- completes the minimum useful tracing story for turn lifecycles
-- keeps long-lived span ownership in the layer that actually owns the turn
-- still builds on the simpler propagation model from PR 2 instead of mixing
-  everything into one change
-
-### PR 4: Optional follow-ups
-
-Possible follow-ups:
-
-1. Reuse existing app-server thread state to add thread loaded/unloaded duration
-   metrics if needed.
-2. Reuse existing realtime runtime state to add realtime duration metrics if
-   needed.
-3. Add outbound JSON-RPC trace propagation only if there is a concrete
-   client-side tracing use case.
-
-## Rollout guidance
-
-Start with:
-
-- inbound request spans for all app-server requests
-- `turn/start` request -> core propagation
-- a core-owned long-lived turn span for `turn/start`
-
-Those pieces exercise the important mechanics:
-
-- inbound carrier extraction
-- request span creation
-- async handoff into core
-- inherited core tracing beneath the propagated parent
-- a single span covering the full duration of a turn
-
-After that, only add more lifecycle-specific tracing if a real debugging or
-observability gap remains.
-
-## Bottom line
-
-The recommended initial design is:
-
-- trace context on inbound JSON-RPC request envelopes
-- one standardized request span for every inbound request
-- automatic propagation through `Submission` into core
-- core-owned long-lived turn spans for turn-producing APIs
-- OTEL conversion and carrier logic centralized in `codex-otel`
-- no app-server-owned tracing registries for turns, threads, or realtime
-  sessions in the initial implementation
-
-This gives app-server distributed tracing that is:
-
-- consistent
-- low-boilerplate
-- modular
-- aligned with the existing ownership boundaries in app-server and core

codex-rs/Cargo.toml

@@ -33,6 +33,8 @@ members = [
     "mcp-server",
     "network-proxy",
     "ollama",
+    "artifact-presentation",
+    "artifact-spreadsheet",
     "process-hardening",
     "protocol",
     "rmcp-client",
@@ -109,6 +111,8 @@ codex-mcp-server = { path = "mcp-server" }
 codex-network-proxy = { path = "network-proxy" }
 codex-ollama = { path = "ollama" }
 codex-otel = { path = "otel" }
+codex-artifact-presentation = { path = "artifact-presentation" }
+codex-artifact-spreadsheet = { path = "artifact-spreadsheet" }
 codex-process-hardening = { path = "process-hardening" }
 codex-protocol = { path = "protocol" }
 codex-responses-api-proxy = { path = "responses-api-proxy" }
@@ -186,6 +190,7 @@ icu_locale_core = "2.1"
 icu_provider = { version = "2.1", features = ["sync"] }
 ignore = "0.4.23"
 image = { version = "^0.25.9", default-features = false }
+iana-time-zone = "0.1.64"
 include_dir = "0.7.4"
 indexmap = "2.12.0"
 insta = "1.46.3"
@@ -214,6 +219,7 @@ owo-colors = "4.3.0"
 path-absolutize = "3.1.1"
 pathdiff = "0.2"
 portable-pty = "0.9.0"
+ppt-rs = "0.2.6"
 predicates = "3"
 pretty_assertions = "1.4.1"
 pulldown-cmark = "0.10"
@@ -348,6 +354,7 @@ ignored = [
     "openssl-sys",
     "codex-utils-readiness",
     "codex-secrets",
+    "codex-artifact-spreadsheet"
 ]
 
 [profile.release]

codex-rs/app-server-protocol/Cargo.toml

@@ -20,6 +20,7 @@ codex-utils-absolute-path = { workspace = true }
 schemars = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
+serde_with = { workspace = true }
 shlex = { workspace = true }
 strum_macros = { workspace = true }
 thiserror = { workspace = true }

codex-rs/app-server-protocol/schema/json/ClientRequest.json

@@ -1340,7 +1340,7 @@
               "type": "string"
             },
             "output": {
-              "type": "string"
+              "$ref": "#/definitions/FunctionCallOutputPayload"
             },
             "type": {
               "enum": [
@@ -1703,6 +1703,12 @@
         }
       ]
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "Settings": {
       "description": "Settings for a collaboration mode.",
       "properties": {
@@ -1933,6 +1939,23 @@
             }
           ]
         },
+        "serviceTier": {
+          "anyOf": [
+            {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ServiceTier"
+                },
+                {
+                  "type": "null"
+                }
+              ]
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
         "threadId": {
           "type": "string"
         }
@@ -2155,6 +2178,23 @@
             }
           ]
         },
+        "serviceTier": {
+          "anyOf": [
+            {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ServiceTier"
+                },
+                {
+                  "type": "null"
+                }
+              ]
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
         "threadId": {
           "type": "string"
         }
@@ -2299,6 +2339,23 @@
             "string",
             "null"
           ]
+        },
+        "serviceTier": {
+          "anyOf": [
+            {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ServiceTier"
+                },
+                {
+                  "type": "null"
+                }
+              ]
+            },
+            {
+              "type": "null"
+            }
+          ]
         }
       },
       "type": "object"
@@ -2409,6 +2466,24 @@
           ],
           "description": "Override the sandbox policy for this turn and subsequent turns."
         },
+        "serviceTier": {
+          "anyOf": [
+            {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ServiceTier"
+                },
+                {
+                  "type": "null"
+                }
+              ]
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "description": "Override the service tier for this turn and subsequent turns."
+        },
         "summary": {
           "anyOf": [
             {

codex-rs/app-server-protocol/schema/json/CommandExecutionRequestApprovalParams.json

@@ -1,11 +1,15 @@
 {
   "$schema": "http://json-schema.org/draft-07/schema#",
   "definitions": {
+    "AbsolutePathBuf": {
+      "description": "A path that is guaranteed to be absolute and normalized (though it is not guaranteed to be canonicalized or exist on the filesystem).\n\nIMPORTANT: When deserializing an `AbsolutePathBuf`, a base path must be set using [AbsolutePathBufGuard::new]. If no base path is set, the deserialization will fail unless the path being deserialized is already absolute.",
+      "type": "string"
+    },
     "AdditionalFileSystemPermissions": {
       "properties": {
         "read": {
           "items": {
-            "type": "string"
+            "$ref": "#/definitions/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -14,7 +18,7 @@
         },
         "write": {
           "items": {
-            "type": "string"
+            "$ref": "#/definitions/AbsolutePathBuf"
           },
           "type": [
             "array",

codex-rs/app-server-protocol/schema/json/EventMsg.json

@@ -1138,6 +1138,16 @@
               ],
               "description": "How to sandbox commands executed in the system"
             },
+            "service_tier": {
+              "anyOf": [
+                {
+                  "$ref": "#/definitions/ServiceTier"
+                },
+                {
+                  "type": "null"
+                }
+              ]
+            },
             "session_id": {
               "$ref": "#/definitions/ThreadId"
             },
@@ -3411,7 +3421,7 @@
       "properties": {
         "read": {
           "items": {
-            "type": "string"
+            "$ref": "#/definitions/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -3420,7 +3430,7 @@
         },
         "write": {
           "items": {
-            "type": "string"
+            "$ref": "#/definitions/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -4234,8 +4244,14 @@
         {
           "additionalProperties": false,
           "properties": {
-            "SessionCreated": {
+            "SessionUpdated": {
               "properties": {
+                "instructions": {
+                  "type": [
+                    "string",
+                    "null"
+                  ]
+                },
                 "session_id": {
                   "type": "string"
                 }
@@ -4246,27 +4262,6 @@
               "type": "object"
             }
           },
-          "required": [
-            "SessionCreated"
-          ],
-          "title": "SessionCreatedRealtimeEvent",
-          "type": "object"
-        },
-        {
-          "additionalProperties": false,
-          "properties": {
-            "SessionUpdated": {
-              "properties": {
-                "backend_prompt": {
-                  "type": [
-                    "string",
-                    "null"
-                  ]
-                }
-              },
-              "type": "object"
-            }
-          },
           "required": [
             "SessionUpdated"
           ],
@@ -4297,6 +4292,40 @@
           "title": "ConversationItemAddedRealtimeEvent",
           "type": "object"
         },
+        {
+          "additionalProperties": false,
+          "properties": {
+            "ConversationItemDone": {
+              "properties": {
+                "item_id": {
+                  "type": "string"
+                }
+              },
+              "required": [
+                "item_id"
+              ],
+              "type": "object"
+            }
+          },
+          "required": [
+            "ConversationItemDone"
+          ],
+          "title": "ConversationItemDoneRealtimeEvent",
+          "type": "object"
+        },
+        {
+          "additionalProperties": false,
+          "properties": {
+            "HandoffRequested": {
+              "$ref": "#/definitions/RealtimeHandoffRequested"
+            }
+          },
+          "required": [
+            "HandoffRequested"
+          ],
+          "title": "HandoffRequestedRealtimeEvent",
+          "type": "object"
+        },
         {
           "additionalProperties": false,
           "properties": {
@@ -4312,6 +4341,47 @@
         }
       ]
     },
+    "RealtimeHandoffMessage": {
+      "properties": {
+        "role": {
+          "type": "string"
+        },
+        "text": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "role",
+        "text"
+      ],
+      "type": "object"
+    },
+    "RealtimeHandoffRequested": {
+      "properties": {
+        "handoff_id": {
+          "type": "string"
+        },
+        "input_transcript": {
+          "type": "string"
+        },
+        "item_id": {
+          "type": "string"
+        },
+        "messages": {
+          "items": {
+            "$ref": "#/definitions/RealtimeHandoffMessage"
+          },
+          "type": "array"
+        }
+      },
+      "required": [
+        "handoff_id",
+        "input_transcript",
+        "item_id",
+        "messages"
+      ],
+      "type": "object"
+    },
     "ReasoningEffort": {
       "description": "See https://platform.openai.com/docs/guides/reasoning?api-mode=responses#get-started-with-reasoning",
       "enum": [
@@ -4822,7 +4892,7 @@
               "type": "string"
             },
             "output": {
-              "type": "string"
+              "$ref": "#/definitions/FunctionCallOutputPayload"
             },
             "type": {
               "enum": [
@@ -5350,6 +5420,12 @@
         }
       ]
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "SessionNetworkProxyRuntime": {
       "properties": {
         "admin_addr": {
@@ -6694,6 +6770,16 @@
           ],
           "description": "How to sandbox commands executed in the system"
         },
+        "service_tier": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
         "session_id": {
           "$ref": "#/definitions/ThreadId"
         },

codex-rs/app-server-protocol/schema/json/JSONRPCMessage.json

@@ -70,7 +70,18 @@
         "method": {
           "type": "string"
         },
-        "params": true
+        "params": true,
+        "trace": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/W3cTraceContext"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "description": "Optional W3C Trace Context for distributed tracing."
+        }
       },
       "required": [
         "id",
@@ -102,6 +113,23 @@
           "type": "integer"
         }
       ]
+    },
+    "W3cTraceContext": {
+      "properties": {
+        "traceparent": {
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "tracestate": {
+          "type": [
+            "string",
+            "null"
+          ]
+        }
+      },
+      "type": "object"
     }
   },
   "description": "Refers to any valid JSON-RPC object that can be decoded off the wire, or encoded to be sent.",

codex-rs/app-server-protocol/schema/json/JSONRPCRequest.json

@@ -11,6 +11,23 @@
           "type": "integer"
         }
       ]
+    },
+    "W3cTraceContext": {
+      "properties": {
+        "traceparent": {
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "tracestate": {
+          "type": [
+            "string",
+            "null"
+          ]
+        }
+      },
+      "type": "object"
     }
   },
   "description": "A request that expects a response.",
@@ -21,7 +38,18 @@
     "method": {
       "type": "string"
     },
-    "params": true
+    "params": true,
+    "trace": {
+      "anyOf": [
+        {
+          "$ref": "#/definitions/W3cTraceContext"
+        },
+        {
+          "type": "null"
+        }
+      ],
+      "description": "Optional W3C Trace Context for distributed tracing."
+    }
   },
   "required": [
     "id",

codex-rs/app-server-protocol/schema/json/ServerNotification.json

@@ -46,6 +46,16 @@
               "type": "null"
             }
           ]
+        },
+        "planType": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/PlanType"
+            },
+            {
+              "type": "null"
+            }
+          ]
         }
       },
       "type": "object"
@@ -1422,6 +1432,32 @@
       ],
       "type": "object"
     },
+    "RequestId": {
+      "anyOf": [
+        {
+          "type": "string"
+        },
+        {
+          "format": "int64",
+          "type": "integer"
+        }
+      ]
+    },
+    "ServerRequestResolvedNotification": {
+      "properties": {
+        "requestId": {
+          "$ref": "#/definitions/RequestId"
+        },
+        "threadId": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "requestId",
+        "threadId"
+      ],
+      "type": "object"
+    },
     "SessionSource": {
       "oneOf": [
         {
@@ -1629,6 +1665,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -1698,6 +1738,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",
@@ -3422,6 +3463,26 @@
       "title": "Item/fileChange/outputDeltaNotification",
       "type": "object"
     },
+    {
+      "properties": {
+        "method": {
+          "enum": [
+            "serverRequest/resolved"
+          ],
+          "title": "ServerRequest/resolvedNotificationMethod",
+          "type": "string"
+        },
+        "params": {
+          "$ref": "#/definitions/ServerRequestResolvedNotification"
+        }
+      },
+      "required": [
+        "method",
+        "params"
+      ],
+      "title": "ServerRequest/resolvedNotification",
+      "type": "object"
+    },
     {
       "properties": {
         "method": {

codex-rs/app-server-protocol/schema/json/ServerRequest.json

@@ -1,11 +1,15 @@
 {
   "$schema": "http://json-schema.org/draft-07/schema#",
   "definitions": {
+    "AbsolutePathBuf": {
+      "description": "A path that is guaranteed to be absolute and normalized (though it is not guaranteed to be canonicalized or exist on the filesystem).\n\nIMPORTANT: When deserializing an `AbsolutePathBuf`, a base path must be set using [AbsolutePathBufGuard::new]. If no base path is set, the deserialization will fail unless the path being deserialized is already absolute.",
+      "type": "string"
+    },
     "AdditionalFileSystemPermissions": {
       "properties": {
         "read": {
           "items": {
-            "type": "string"
+            "$ref": "#/definitions/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -14,7 +18,7 @@
         },
         "write": {
           "items": {
-            "type": "string"
+            "$ref": "#/definitions/AbsolutePathBuf"
           },
           "type": [
             "array",

codex-rs/app-server-protocol/schema/json/v2/AccountUpdatedNotification.json

@@ -26,6 +26,20 @@
           "type": "string"
         }
       ]
+    },
+    "PlanType": {
+      "enum": [
+        "free",
+        "go",
+        "plus",
+        "pro",
+        "team",
+        "business",
+        "enterprise",
+        "edu",
+        "unknown"
+      ],
+      "type": "string"
     }
   },
   "properties": {
@@ -38,6 +52,16 @@
           "type": "null"
         }
       ]
+    },
+    "planType": {
+      "anyOf": [
+        {
+          "$ref": "#/definitions/PlanType"
+        },
+        {
+          "type": "null"
+        }
+      ]
     }
   },
   "title": "AccountUpdatedNotification",

codex-rs/app-server-protocol/schema/json/v2/ConfigReadResponse.json

@@ -323,6 +323,16 @@
             }
           ]
         },
+        "service_tier": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
         "tools": {
           "anyOf": [
             {
@@ -608,6 +618,16 @@
             }
           ]
         },
+        "service_tier": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
         "web_search": {
           "anyOf": [
             {
@@ -685,6 +705,12 @@
       },
       "type": "object"
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "ToolsV2": {
       "properties": {
         "view_image": {

codex-rs/app-server-protocol/schema/json/v2/ModelListResponse.json

@@ -22,6 +22,16 @@
     },
     "Model": {
       "properties": {
+        "availabilityNux": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ModelAvailabilityNux"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
         "defaultReasoningEffort": {
           "$ref": "#/definitions/ReasoningEffort"
         },
@@ -68,6 +78,16 @@
             "string",
             "null"
           ]
+        },
+        "upgradeInfo": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ModelUpgradeInfo"
+            },
+            {
+              "type": "null"
+            }
+          ]
         }
       },
       "required": [
@@ -82,6 +102,46 @@
       ],
       "type": "object"
     },
+    "ModelAvailabilityNux": {
+      "properties": {
+        "message": {
+          "type": "string"
+        }
+      },
+      "required": [
+        "message"
+      ],
+      "type": "object"
+    },
+    "ModelUpgradeInfo": {
+      "properties": {
+        "migrationMarkdown": {
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "model": {
+          "type": "string"
+        },
+        "modelLink": {
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "upgradeCopy": {
+          "type": [
+            "string",
+            "null"
+          ]
+        }
+      },
+      "required": [
+        "model"
+      ],
+      "type": "object"
+    },
     "ReasoningEffort": {
       "description": "See https://platform.openai.com/docs/guides/reasoning?api-mode=responses#get-started-with-reasoning",
       "enum": [

codex-rs/app-server-protocol/schema/json/v2/RawResponseItemCompletedNotification.json

@@ -565,7 +565,7 @@
               "type": "string"
             },
             "output": {
-              "type": "string"
+              "$ref": "#/definitions/FunctionCallOutputPayload"
             },
             "type": {
               "enum": [

codex-rs/app-server-protocol/schema/json/v2/ServerRequestResolvedNotification.json

@@ -0,0 +1,30 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "definitions": {
+    "RequestId": {
+      "anyOf": [
+        {
+          "type": "string"
+        },
+        {
+          "format": "int64",
+          "type": "integer"
+        }
+      ]
+    }
+  },
+  "properties": {
+    "requestId": {
+      "$ref": "#/definitions/RequestId"
+    },
+    "threadId": {
+      "type": "string"
+    }
+  },
+  "required": [
+    "requestId",
+    "threadId"
+  ],
+  "title": "ServerRequestResolvedNotification",
+  "type": "object"
+}

codex-rs/app-server-protocol/schema/json/v2/ThreadForkParams.json

@@ -50,6 +50,12 @@
         "danger-full-access"
       ],
       "type": "string"
+    },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
     }
   },
   "description": "There are two ways to fork a thread: 1. By thread_id: load the thread from disk by thread_id and fork it into a new thread. 2. By path: load the thread from disk by path and fork it into a new thread.\n\nIf using path, the thread_id param will be ignored.\n\nPrefer using thread_id whenever possible.",
@@ -112,6 +118,23 @@
         }
       ]
     },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        {
+          "type": "null"
+        }
+      ]
+    },
     "threadId": {
       "type": "string"
     }

codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json

@@ -738,6 +738,12 @@
         }
       ]
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "SessionSource": {
       "oneOf": [
         {
@@ -882,6 +888,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -951,6 +961,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",
@@ -1901,6 +1912,16 @@
     "sandbox": {
       "$ref": "#/definitions/SandboxPolicy"
     },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "$ref": "#/definitions/ServiceTier"
+        },
+        {
+          "type": "null"
+        }
+      ]
+    },
     "thread": {
       "$ref": "#/definitions/Thread"
     }

codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json

@@ -655,6 +655,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -724,6 +728,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",

codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json

@@ -655,6 +655,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -724,6 +728,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",

codex-rs/app-server-protocol/schema/json/v2/ThreadResumeParams.json

@@ -615,7 +615,7 @@
               "type": "string"
             },
             "output": {
-              "type": "string"
+              "$ref": "#/definitions/FunctionCallOutputPayload"
             },
             "type": {
               "enum": [
@@ -738,6 +738,12 @@
       ],
       "type": "string"
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "WebSearchAction": {
       "oneOf": [
         {
@@ -910,6 +916,23 @@
         }
       ]
     },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        {
+          "type": "null"
+        }
+      ]
+    },
     "threadId": {
       "type": "string"
     }

codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json

@@ -738,6 +738,12 @@
         }
       ]
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "SessionSource": {
       "oneOf": [
         {
@@ -882,6 +888,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -951,6 +961,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",
@@ -1901,6 +1912,16 @@
     "sandbox": {
       "$ref": "#/definitions/SandboxPolicy"
     },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "$ref": "#/definitions/ServiceTier"
+        },
+        {
+          "type": "null"
+        }
+      ]
+    },
     "thread": {
       "$ref": "#/definitions/Thread"
     }

codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json

@@ -655,6 +655,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -724,6 +728,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",

codex-rs/app-server-protocol/schema/json/v2/ThreadStartParams.json

@@ -75,6 +75,12 @@
         "danger-full-access"
       ],
       "type": "string"
+    },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
     }
   },
   "properties": {
@@ -156,6 +162,23 @@
         "string",
         "null"
       ]
+    },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        {
+          "type": "null"
+        }
+      ]
     }
   },
   "title": "ThreadStartParams",

codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json

@@ -738,6 +738,12 @@
         }
       ]
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "SessionSource": {
       "oneOf": [
         {
@@ -882,6 +888,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -951,6 +961,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",
@@ -1901,6 +1912,16 @@
     "sandbox": {
       "$ref": "#/definitions/SandboxPolicy"
     },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "$ref": "#/definitions/ServiceTier"
+        },
+        {
+          "type": "null"
+        }
+      ]
+    },
     "thread": {
       "$ref": "#/definitions/Thread"
     }

codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json

@@ -655,6 +655,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -724,6 +728,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",

codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json

@@ -655,6 +655,10 @@
           "description": "Working directory captured for the thread.",
           "type": "string"
         },
+        "ephemeral": {
+          "description": "Whether the thread is ephemeral and should not be materialized on disk.",
+          "type": "boolean"
+        },
         "gitInfo": {
           "anyOf": [
             {
@@ -724,6 +728,7 @@
         "cliVersion",
         "createdAt",
         "cwd",
+        "ephemeral",
         "id",
         "modelProvider",
         "preview",

codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json

@@ -299,6 +299,12 @@
         }
       ]
     },
+    "ServiceTier": {
+      "enum": [
+        "fast"
+      ],
+      "type": "string"
+    },
     "Settings": {
       "description": "Settings for a collaboration mode.",
       "properties": {
@@ -539,6 +545,24 @@
       ],
       "description": "Override the sandbox policy for this turn and subsequent turns."
     },
+    "serviceTier": {
+      "anyOf": [
+        {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/ServiceTier"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        {
+          "type": "null"
+        }
+      ],
+      "description": "Override the service tier for this turn and subsequent turns."
+    },
     "summary": {
       "anyOf": [
         {

codex-rs/app-server-protocol/schema/typescript/FileSystemPermissions.ts

@@ -1,5 +1,6 @@
 // GENERATED CODE! DO NOT MODIFY BY HAND!
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { AbsolutePathBuf } from "./AbsolutePathBuf";
 
-export type FileSystemPermissions = { read: Array<string> | null, write: Array<string> | null, };
+export type FileSystemPermissions = { read: Array<AbsolutePathBuf> | null, write: Array<AbsolutePathBuf> | null, };

codex-rs/app-server-protocol/schema/typescript/RealtimeEvent.ts

@@ -2,6 +2,7 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { RealtimeAudioFrame } from "./RealtimeAudioFrame";
+import type { RealtimeHandoffRequested } from "./RealtimeHandoffRequested";
 import type { JsonValue } from "./serde_json/JsonValue";
 
-export type RealtimeEvent = { "SessionCreated": { session_id: string, } } | { "SessionUpdated": { backend_prompt: string | null, } } | { "AudioOut": RealtimeAudioFrame } | { "ConversationItemAdded": JsonValue } | { "Error": string };
+export type RealtimeEvent = { "SessionUpdated": { session_id: string, instructions: string | null, } } | { "AudioOut": RealtimeAudioFrame } | { "ConversationItemAdded": JsonValue } | { "ConversationItemDone": { item_id: string, } } | { "HandoffRequested": RealtimeHandoffRequested } | { "Error": string };

codex-rs/app-server-protocol/schema/typescript/RealtimeHandoffMessage.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type RealtimeHandoffMessage = { role: string, text: string, };

codex-rs/app-server-protocol/schema/typescript/RealtimeHandoffRequested.ts

@@ -0,0 +1,6 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { RealtimeHandoffMessage } from "./RealtimeHandoffMessage";
+
+export type RealtimeHandoffRequested = { handoff_id: string, item_id: string, input_transcript: string, messages: Array<RealtimeHandoffMessage>, };

codex-rs/app-server-protocol/schema/typescript/ResponseItem.ts

@@ -15,4 +15,4 @@ export type ResponseItem = { "type": "message", role: string, content: Array<Con
 /**
  * Set when using the Responses API.
  */
-call_id: string | null, status: LocalShellStatus, action: LocalShellAction, } | { "type": "function_call", name: string, arguments: string, call_id: string, } | { "type": "function_call_output", call_id: string, output: FunctionCallOutputPayload, } | { "type": "custom_tool_call", status?: string, call_id: string, name: string, input: string, } | { "type": "custom_tool_call_output", call_id: string, output: string, } | { "type": "web_search_call", status?: string, action?: WebSearchAction, } | { "type": "ghost_snapshot", ghost_commit: GhostCommit, } | { "type": "compaction", encrypted_content: string, } | { "type": "other" };
+call_id: string | null, status: LocalShellStatus, action: LocalShellAction, } | { "type": "function_call", name: string, arguments: string, call_id: string, } | { "type": "function_call_output", call_id: string, output: FunctionCallOutputPayload, } | { "type": "custom_tool_call", status?: string, call_id: string, name: string, input: string, } | { "type": "custom_tool_call_output", call_id: string, output: FunctionCallOutputPayload, } | { "type": "web_search_call", status?: string, action?: WebSearchAction, } | { "type": "ghost_snapshot", ghost_commit: GhostCommit, } | { "type": "compaction", encrypted_content: string, } | { "type": "other" };

codex-rs/app-server-protocol/schema/typescript/SendUserTurnParams.ts

@@ -6,10 +6,11 @@ import type { InputItem } from "./InputItem";
 import type { ReasoningEffort } from "./ReasoningEffort";
 import type { ReasoningSummary } from "./ReasoningSummary";
 import type { SandboxPolicy } from "./SandboxPolicy";
+import type { ServiceTier } from "./ServiceTier";
 import type { ThreadId } from "./ThreadId";
 import type { JsonValue } from "./serde_json/JsonValue";
 
-export type SendUserTurnParams = { conversationId: ThreadId, items: Array<InputItem>, cwd: string, approvalPolicy: AskForApproval, sandboxPolicy: SandboxPolicy, model: string, effort: ReasoningEffort | null, summary: ReasoningSummary, 
+export type SendUserTurnParams = { conversationId: ThreadId, items: Array<InputItem>, cwd: string, approvalPolicy: AskForApproval, sandboxPolicy: SandboxPolicy, model: string, serviceTier?: ServiceTier | null | null, effort: ReasoningEffort | null, summary: ReasoningSummary, 
 /**
  * Optional JSON Schema used to constrain the final assistant message for this turn.
  */

codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts

@@ -27,6 +27,7 @@ import type { RawResponseItemCompletedNotification } from "./v2/RawResponseItemC
 import type { ReasoningSummaryPartAddedNotification } from "./v2/ReasoningSummaryPartAddedNotification";
 import type { ReasoningSummaryTextDeltaNotification } from "./v2/ReasoningSummaryTextDeltaNotification";
 import type { ReasoningTextDeltaNotification } from "./v2/ReasoningTextDeltaNotification";
+import type { ServerRequestResolvedNotification } from "./v2/ServerRequestResolvedNotification";
 import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNotification";
 import type { ThreadArchivedNotification } from "./v2/ThreadArchivedNotification";
 import type { ThreadClosedNotification } from "./v2/ThreadClosedNotification";
@@ -50,4 +51,4 @@ import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldW
 /**
  * Notification sent from the server to the client.
  */
-export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification } | { "method": "authStatusChange", "params": AuthStatusChangeNotification } | { "method": "loginChatGptComplete", "params": LoginChatGptCompleteNotification } | { "method": "sessionConfigured", "params": SessionConfiguredNotification };
+export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification } | { "method": "authStatusChange", "params": AuthStatusChangeNotification } | { "method": "loginChatGptComplete", "params": LoginChatGptCompleteNotification } | { "method": "sessionConfigured", "params": SessionConfiguredNotification };

codex-rs/app-server-protocol/schema/typescript/ServiceTier.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type ServiceTier = "fast";

codex-rs/app-server-protocol/schema/typescript/SessionConfiguredEvent.ts

@@ -5,6 +5,7 @@ import type { AskForApproval } from "./AskForApproval";
 import type { EventMsg } from "./EventMsg";
 import type { ReasoningEffort } from "./ReasoningEffort";
 import type { SandboxPolicy } from "./SandboxPolicy";
+import type { ServiceTier } from "./ServiceTier";
 import type { SessionNetworkProxyRuntime } from "./SessionNetworkProxyRuntime";
 import type { ThreadId } from "./ThreadId";
 
@@ -16,7 +17,7 @@ thread_name?: string,
 /**
  * Tell the client what model is being queried.
  */
-model: string, model_provider_id: string, 
+model: string, model_provider_id: string, service_tier: ServiceTier | null, 
 /**
  * When to escalate for approval for execution
  */

codex-rs/app-server-protocol/schema/typescript/SessionConfiguredNotification.ts

@@ -3,6 +3,7 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { EventMsg } from "./EventMsg";
 import type { ReasoningEffort } from "./ReasoningEffort";
+import type { ServiceTier } from "./ServiceTier";
 import type { ThreadId } from "./ThreadId";
 
-export type SessionConfiguredNotification = { sessionId: ThreadId, model: string, reasoningEffort: ReasoningEffort | null, historyLogId: bigint, historyEntryCount: number, initialMessages: Array<EventMsg> | null, rolloutPath: string, };
+export type SessionConfiguredNotification = { sessionId: ThreadId, model: string, serviceTier: ServiceTier | null, reasoningEffort: ReasoningEffort | null, historyLogId: bigint, historyEntryCount: number, initialMessages: Array<EventMsg> | null, rolloutPath: string, };

codex-rs/app-server-protocol/schema/typescript/index.ts

@@ -161,6 +161,8 @@ export type { RealtimeConversationClosedEvent } from "./RealtimeConversationClos
 export type { RealtimeConversationRealtimeEvent } from "./RealtimeConversationRealtimeEvent";
 export type { RealtimeConversationStartedEvent } from "./RealtimeConversationStartedEvent";
 export type { RealtimeEvent } from "./RealtimeEvent";
+export type { RealtimeHandoffMessage } from "./RealtimeHandoffMessage";
+export type { RealtimeHandoffRequested } from "./RealtimeHandoffRequested";
 export type { ReasoningContentDeltaEvent } from "./ReasoningContentDeltaEvent";
 export type { ReasoningEffort } from "./ReasoningEffort";
 export type { ReasoningItem } from "./ReasoningItem";
@@ -198,6 +200,7 @@ export type { SendUserTurnParams } from "./SendUserTurnParams";
 export type { SendUserTurnResponse } from "./SendUserTurnResponse";
 export type { ServerNotification } from "./ServerNotification";
 export type { ServerRequest } from "./ServerRequest";
+export type { ServiceTier } from "./ServiceTier";
 export type { SessionConfiguredEvent } from "./SessionConfiguredEvent";
 export type { SessionConfiguredNotification } from "./SessionConfiguredNotification";
 export type { SessionNetworkProxyRuntime } from "./SessionNetworkProxyRuntime";

codex-rs/app-server-protocol/schema/typescript/v2/AccountUpdatedNotification.ts

@@ -2,5 +2,6 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { AuthMode } from "../AuthMode";
+import type { PlanType } from "../PlanType";
 
-export type AccountUpdatedNotification = { authMode: AuthMode | null, };
+export type AccountUpdatedNotification = { authMode: AuthMode | null, planType: PlanType | null, };

codex-rs/app-server-protocol/schema/typescript/v2/AdditionalFileSystemPermissions.ts

@@ -1,5 +1,6 @@
 // GENERATED CODE! DO NOT MODIFY BY HAND!
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { AbsolutePathBuf } from "../AbsolutePathBuf";
 
-export type AdditionalFileSystemPermissions = { read: Array<string> | null, write: Array<string> | null, };
+export type AdditionalFileSystemPermissions = { read: Array<AbsolutePathBuf> | null, write: Array<AbsolutePathBuf> | null, };

codex-rs/app-server-protocol/schema/typescript/v2/Config.ts

@@ -4,6 +4,7 @@
 import type { ForcedLoginMethod } from "../ForcedLoginMethod";
 import type { ReasoningEffort } from "../ReasoningEffort";
 import type { ReasoningSummary } from "../ReasoningSummary";
+import type { ServiceTier } from "../ServiceTier";
 import type { Verbosity } from "../Verbosity";
 import type { WebSearchMode } from "../WebSearchMode";
 import type { JsonValue } from "../serde_json/JsonValue";
@@ -14,4 +15,4 @@ import type { SandboxMode } from "./SandboxMode";
 import type { SandboxWorkspaceWrite } from "./SandboxWorkspaceWrite";
 import type { ToolsV2 } from "./ToolsV2";
 
-export type Config = {model: string | null, review_model: string | null, model_context_window: bigint | null, model_auto_compact_token_limit: bigint | null, model_provider: string | null, approval_policy: AskForApproval | null, sandbox_mode: SandboxMode | null, sandbox_workspace_write: SandboxWorkspaceWrite | null, forced_chatgpt_workspace_id: string | null, forced_login_method: ForcedLoginMethod | null, web_search: WebSearchMode | null, tools: ToolsV2 | null, profile: string | null, profiles: { [key in string]?: ProfileV2 }, instructions: string | null, developer_instructions: string | null, compact_prompt: string | null, model_reasoning_effort: ReasoningEffort | null, model_reasoning_summary: ReasoningSummary | null, model_verbosity: Verbosity | null, analytics: AnalyticsConfig | null} & ({ [key in string]?: number | string | boolean | Array<JsonValue> | { [key in string]?: JsonValue } | null });
+export type Config = {model: string | null, review_model: string | null, model_context_window: bigint | null, model_auto_compact_token_limit: bigint | null, model_provider: string | null, approval_policy: AskForApproval | null, sandbox_mode: SandboxMode | null, sandbox_workspace_write: SandboxWorkspaceWrite | null, forced_chatgpt_workspace_id: string | null, forced_login_method: ForcedLoginMethod | null, web_search: WebSearchMode | null, tools: ToolsV2 | null, profile: string | null, profiles: { [key in string]?: ProfileV2 }, instructions: string | null, developer_instructions: string | null, compact_prompt: string | null, model_reasoning_effort: ReasoningEffort | null, model_reasoning_summary: ReasoningSummary | null, model_verbosity: Verbosity | null, service_tier: ServiceTier | null, analytics: AnalyticsConfig | null} & ({ [key in string]?: number | string | boolean | Array<JsonValue> | { [key in string]?: JsonValue } | null });

codex-rs/app-server-protocol/schema/typescript/v2/Model.ts

@@ -3,6 +3,8 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { InputModality } from "../InputModality";
 import type { ReasoningEffort } from "../ReasoningEffort";
+import type { ModelAvailabilityNux } from "./ModelAvailabilityNux";
+import type { ModelUpgradeInfo } from "./ModelUpgradeInfo";
 import type { ReasoningEffortOption } from "./ReasoningEffortOption";
 
-export type Model = { id: string, model: string, upgrade: string | null, displayName: string, description: string, hidden: boolean, supportedReasoningEfforts: Array<ReasoningEffortOption>, defaultReasoningEffort: ReasoningEffort, inputModalities: Array<InputModality>, supportsPersonality: boolean, isDefault: boolean, };
+export type Model = { id: string, model: string, upgrade: string | null, upgradeInfo: ModelUpgradeInfo | null, availabilityNux: ModelAvailabilityNux | null, displayName: string, description: string, hidden: boolean, supportedReasoningEfforts: Array<ReasoningEffortOption>, defaultReasoningEffort: ReasoningEffort, inputModalities: Array<InputModality>, supportsPersonality: boolean, isDefault: boolean, };

codex-rs/app-server-protocol/schema/typescript/v2/ModelAvailabilityNux.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type ModelAvailabilityNux = { message: string, };

codex-rs/app-server-protocol/schema/typescript/v2/ModelUpgradeInfo.ts

@@ -0,0 +1,5 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+
+export type ModelUpgradeInfo = { model: string, upgradeCopy: string | null, modelLink: string | null, migrationMarkdown: string | null, };

codex-rs/app-server-protocol/schema/typescript/v2/ProfileV2.ts

@@ -3,9 +3,10 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { ReasoningEffort } from "../ReasoningEffort";
 import type { ReasoningSummary } from "../ReasoningSummary";
+import type { ServiceTier } from "../ServiceTier";
 import type { Verbosity } from "../Verbosity";
 import type { WebSearchMode } from "../WebSearchMode";
 import type { JsonValue } from "../serde_json/JsonValue";
 import type { AskForApproval } from "./AskForApproval";
 
-export type ProfileV2 = { model: string | null, model_provider: string | null, approval_policy: AskForApproval | null, model_reasoning_effort: ReasoningEffort | null, model_reasoning_summary: ReasoningSummary | null, model_verbosity: Verbosity | null, web_search: WebSearchMode | null, chatgpt_base_url: string | null, } & ({ [key in string]?: number | string | boolean | Array<JsonValue> | { [key in string]?: JsonValue } | null });
+export type ProfileV2 = { model: string | null, model_provider: string | null, approval_policy: AskForApproval | null, service_tier: ServiceTier | null, model_reasoning_effort: ReasoningEffort | null, model_reasoning_summary: ReasoningSummary | null, model_verbosity: Verbosity | null, web_search: WebSearchMode | null, chatgpt_base_url: string | null, } & ({ [key in string]?: number | string | boolean | Array<JsonValue> | { [key in string]?: JsonValue } | null });

codex-rs/app-server-protocol/schema/typescript/v2/ServerRequestResolvedNotification.ts

@@ -0,0 +1,6 @@
+// GENERATED CODE! DO NOT MODIFY BY HAND!
+
+// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { RequestId } from "../RequestId";
+
+export type ServerRequestResolvedNotification = { threadId: string, requestId: RequestId, };

codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts

@@ -11,6 +11,10 @@ export type Thread = { id: string,
  * Usually the first user message in the thread, if available.
  */
 preview: string, 
+/**
+ * Whether the thread is ephemeral and should not be materialized on disk.
+ */
+ephemeral: boolean, 
 /**
  * Model provider used for this thread (for example, 'openai').
  */

codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts

@@ -1,6 +1,7 @@
 // GENERATED CODE! DO NOT MODIFY BY HAND!
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
+import type { ServiceTier } from "../ServiceTier";
 import type { JsonValue } from "../serde_json/JsonValue";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxMode } from "./SandboxMode";
@@ -21,7 +22,7 @@ export type ThreadForkParams = {threadId: string, /**
 path?: string | null, /**
  * Configuration overrides for the forked thread, if any.
  */
-model?: string | null, modelProvider?: string | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, /**
+model?: string | null, modelProvider?: string | null, serviceTier?: ServiceTier | null | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, /**
  * If true, persist additional rollout EventMsg variants required to
  * reconstruct a richer thread history on subsequent resume/fork/read.
  */

codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkResponse.ts

@@ -2,8 +2,9 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { ReasoningEffort } from "../ReasoningEffort";
+import type { ServiceTier } from "../ServiceTier";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxPolicy } from "./SandboxPolicy";
 import type { Thread } from "./Thread";
 
-export type ThreadForkResponse = { thread: Thread, model: string, modelProvider: string, cwd: string, approvalPolicy: AskForApproval, sandbox: SandboxPolicy, reasoningEffort: ReasoningEffort | null, };
+export type ThreadForkResponse = { thread: Thread, model: string, modelProvider: string, serviceTier: ServiceTier | null, cwd: string, approvalPolicy: AskForApproval, sandbox: SandboxPolicy, reasoningEffort: ReasoningEffort | null, };

codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeParams.ts

@@ -3,6 +3,7 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { Personality } from "../Personality";
 import type { ResponseItem } from "../ResponseItem";
+import type { ServiceTier } from "../ServiceTier";
 import type { JsonValue } from "../serde_json/JsonValue";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxMode } from "./SandboxMode";
@@ -30,7 +31,7 @@ history?: Array<ResponseItem> | null, /**
 path?: string | null, /**
  * Configuration overrides for the resumed thread, if any.
  */
-model?: string | null, modelProvider?: string | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, /**
+model?: string | null, modelProvider?: string | null, serviceTier?: ServiceTier | null | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, /**
  * If true, persist additional rollout EventMsg variants required to
  * reconstruct a richer thread history on subsequent resume/fork/read.
  */

codex-rs/app-server-protocol/schema/typescript/v2/ThreadResumeResponse.ts

@@ -2,8 +2,9 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { ReasoningEffort } from "../ReasoningEffort";
+import type { ServiceTier } from "../ServiceTier";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxPolicy } from "./SandboxPolicy";
 import type { Thread } from "./Thread";
 
-export type ThreadResumeResponse = { thread: Thread, model: string, modelProvider: string, cwd: string, approvalPolicy: AskForApproval, sandbox: SandboxPolicy, reasoningEffort: ReasoningEffort | null, };
+export type ThreadResumeResponse = { thread: Thread, model: string, modelProvider: string, serviceTier: ServiceTier | null, cwd: string, approvalPolicy: AskForApproval, sandbox: SandboxPolicy, reasoningEffort: ReasoningEffort | null, };

codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartParams.ts

@@ -2,11 +2,12 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { Personality } from "../Personality";
+import type { ServiceTier } from "../ServiceTier";
 import type { JsonValue } from "../serde_json/JsonValue";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxMode } from "./SandboxMode";
 
-export type ThreadStartParams = {model?: string | null, modelProvider?: string | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, serviceName?: string | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, ephemeral?: boolean | null, /**
+export type ThreadStartParams = {model?: string | null, modelProvider?: string | null, serviceTier?: ServiceTier | null | null, cwd?: string | null, approvalPolicy?: AskForApproval | null, sandbox?: SandboxMode | null, config?: { [key in string]?: JsonValue } | null, serviceName?: string | null, baseInstructions?: string | null, developerInstructions?: string | null, personality?: Personality | null, ephemeral?: boolean | null, /**
  * If true, opt into emitting raw Responses API items on the event stream.
  * This is for internal use only (e.g. Codex Cloud).
  */

codex-rs/app-server-protocol/schema/typescript/v2/ThreadStartResponse.ts

@@ -2,8 +2,9 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { ReasoningEffort } from "../ReasoningEffort";
+import type { ServiceTier } from "../ServiceTier";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxPolicy } from "./SandboxPolicy";
 import type { Thread } from "./Thread";
 
-export type ThreadStartResponse = { thread: Thread, model: string, modelProvider: string, cwd: string, approvalPolicy: AskForApproval, sandbox: SandboxPolicy, reasoningEffort: ReasoningEffort | null, };
+export type ThreadStartResponse = { thread: Thread, model: string, modelProvider: string, serviceTier: ServiceTier | null, cwd: string, approvalPolicy: AskForApproval, sandbox: SandboxPolicy, reasoningEffort: ReasoningEffort | null, };

codex-rs/app-server-protocol/schema/typescript/v2/TurnStartParams.ts

@@ -5,6 +5,7 @@ import type { CollaborationMode } from "../CollaborationMode";
 import type { Personality } from "../Personality";
 import type { ReasoningEffort } from "../ReasoningEffort";
 import type { ReasoningSummary } from "../ReasoningSummary";
+import type { ServiceTier } from "../ServiceTier";
 import type { JsonValue } from "../serde_json/JsonValue";
 import type { AskForApproval } from "./AskForApproval";
 import type { SandboxPolicy } from "./SandboxPolicy";
@@ -23,6 +24,9 @@ sandboxPolicy?: SandboxPolicy | null, /**
  * Override the model for this turn and subsequent turns.
  */
 model?: string | null, /**
+ * Override the service tier for this turn and subsequent turns.
+ */
+serviceTier?: ServiceTier | null | null, /**
  * Override the reasoning effort for this turn and subsequent turns.
  */
 effort?: ReasoningEffort | null, /**

codex-rs/app-server-protocol/schema/typescript/v2/index.ts

@@ -107,10 +107,12 @@ export type { McpToolCallResult } from "./McpToolCallResult";
 export type { McpToolCallStatus } from "./McpToolCallStatus";
 export type { MergeStrategy } from "./MergeStrategy";
 export type { Model } from "./Model";
+export type { ModelAvailabilityNux } from "./ModelAvailabilityNux";
 export type { ModelListParams } from "./ModelListParams";
 export type { ModelListResponse } from "./ModelListResponse";
 export type { ModelRerouteReason } from "./ModelRerouteReason";
 export type { ModelReroutedNotification } from "./ModelReroutedNotification";
+export type { ModelUpgradeInfo } from "./ModelUpgradeInfo";
 export type { NetworkAccess } from "./NetworkAccess";
 export type { NetworkApprovalContext } from "./NetworkApprovalContext";
 export type { NetworkApprovalProtocol } from "./NetworkApprovalProtocol";
@@ -140,6 +142,7 @@ export type { ReviewTarget } from "./ReviewTarget";
 export type { SandboxMode } from "./SandboxMode";
 export type { SandboxPolicy } from "./SandboxPolicy";
 export type { SandboxWorkspaceWrite } from "./SandboxWorkspaceWrite";
+export type { ServerRequestResolvedNotification } from "./ServerRequestResolvedNotification";
 export type { SessionSource } from "./SessionSource";
 export type { SkillDependencies } from "./SkillDependencies";
 export type { SkillErrorInfo } from "./SkillErrorInfo";

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

@@ -1,6 +1,7 @@
 //! We do not do true JSON-RPC 2.0, as we neither send nor expect the
 //! "jsonrpc": "2.0" field.
 
+use codex_protocol::protocol::W3cTraceContext;
 use schemars::JsonSchema;
 use serde::Deserialize;
 use serde::Serialize;
@@ -8,7 +9,9 @@ use ts_rs::TS;
 
 pub const JSONRPC_VERSION: &str = "2.0";
 
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Hash, Eq, JsonSchema, TS)]
+#[derive(
+    Debug, Clone, PartialEq, PartialOrd, Ord, Deserialize, Serialize, Hash, Eq, JsonSchema, TS,
+)]
 #[serde(untagged)]
 pub enum RequestId {
     String(String),
@@ -36,6 +39,10 @@ pub struct JSONRPCRequest {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     #[ts(optional)]
     pub params: Option<serde_json::Value>,
+    /// Optional W3C Trace Context for distributed tracing.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[ts(optional)]
+    pub trace: Option<W3cTraceContext>,
 }
 
 /// A notification which does not expect a response.

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

@@ -548,6 +548,14 @@ macro_rules! server_request_definitions {
             )*
         }
 
+        impl ServerRequest {
+            pub fn id(&self) -> &RequestId {
+                match self {
+                    $(Self::$variant { request_id, .. } => request_id,)*
+                }
+            }
+        }
+
         #[derive(Debug, Clone, PartialEq, JsonSchema)]
         #[allow(clippy::large_enum_variant)]
         pub enum ServerRequestPayload {
@@ -838,6 +846,7 @@ server_notification_definitions! {
     CommandExecutionOutputDelta => "item/commandExecution/outputDelta" (v2::CommandExecutionOutputDeltaNotification),
     TerminalInteraction => "item/commandExecution/terminalInteraction" (v2::TerminalInteractionNotification),
     FileChangeOutputDelta => "item/fileChange/outputDelta" (v2::FileChangeOutputDeltaNotification),
+    ServerRequestResolved => "serverRequest/resolved" (v2::ServerRequestResolvedNotification),
     McpToolCallProgress => "item/mcpToolCall/progress" (v2::McpToolCallProgressNotification),
     McpServerOauthLoginCompleted => "mcpServer/oauthLogin/completed" (v2::McpServerOauthLoginCompletedNotification),
     AccountUpdated => "account/updated" (v2::AccountUpdatedNotification),
@@ -893,10 +902,15 @@ mod tests {
     use codex_protocol::account::PlanType;
     use codex_protocol::parse_command::ParsedCommand;
     use codex_protocol::protocol::AskForApproval;
+    use codex_utils_absolute_path::AbsolutePathBuf;
     use pretty_assertions::assert_eq;
     use serde_json::json;
     use std::path::PathBuf;
 
+    fn absolute_path(path: &str) -> AbsolutePathBuf {
+        AbsolutePathBuf::from_absolute_path(path).expect("absolute path")
+    }
+
     #[test]
     fn serialize_new_conversation() -> Result<()> {
         let request = ClientRequest::NewConversation {
@@ -1101,6 +1115,7 @@ mod tests {
         );
 
         let payload = ServerRequestPayload::ExecCommandApproval(params);
+        assert_eq!(request.id(), &RequestId::Integer(7));
         assert_eq!(payload.request_with_id(RequestId::Integer(7)), request);
         Ok(())
     }
@@ -1533,7 +1548,7 @@ mod tests {
             additional_permissions: Some(v2::AdditionalPermissionProfile {
                 network: None,
                 file_system: Some(v2::AdditionalFileSystemPermissions {
-                    read: Some(vec![std::path::PathBuf::from("/tmp/allowed")]),
+                    read: Some(vec![absolute_path("/tmp/allowed")]),
                     write: None,
                 }),
                 macos: None,

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

@@ -3,6 +3,7 @@
 
 pub mod common;
 mod mappers;
+mod serde_helpers;
 pub mod thread_history;
 pub mod v1;
 pub mod v2;

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

@@ -0,0 +1,23 @@
+use serde::Deserialize;
+use serde::Deserializer;
+use serde::Serialize;
+use serde::Serializer;
+
+pub fn deserialize_double_option<'de, T, D>(deserializer: D) -> Result<Option<Option<T>>, D::Error>
+where
+    T: Deserialize<'de>,
+    D: Deserializer<'de>,
+{
+    serde_with::rust::double_option::deserialize(deserializer)
+}
+
+pub fn serialize_double_option<T, S>(
+    value: &Option<Option<T>>,
+    serializer: S,
+) -> Result<S::Ok, S::Error>
+where
+    T: Serialize,
+    S: Serializer,
+{
+    serde_with::rust::double_option::serialize(value, serializer)
+}

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

@@ -22,6 +22,7 @@ use codex_protocol::models::MessagePhase;
 use codex_protocol::protocol::AgentReasoningEvent;
 use codex_protocol::protocol::AgentReasoningRawContentEvent;
 use codex_protocol::protocol::AgentStatus;
+use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
 use codex_protocol::protocol::CompactedItem;
 use codex_protocol::protocol::ContextCompactedEvent;
 use codex_protocol::protocol::DynamicToolCallResponseEvent;
@@ -126,6 +127,9 @@ impl ThreadHistoryBuilder {
             EventMsg::WebSearchEnd(payload) => self.handle_web_search_end(payload),
             EventMsg::ExecCommandBegin(payload) => self.handle_exec_command_begin(payload),
             EventMsg::ExecCommandEnd(payload) => self.handle_exec_command_end(payload),
+            EventMsg::ApplyPatchApprovalRequest(payload) => {
+                self.handle_apply_patch_approval_request(payload)
+            }
             EventMsg::PatchApplyBegin(payload) => self.handle_patch_apply_begin(payload),
             EventMsg::PatchApplyEnd(payload) => self.handle_patch_apply_end(payload),
             EventMsg::DynamicToolCallRequest(payload) => {
@@ -364,6 +368,19 @@ impl ThreadHistoryBuilder {
         self.upsert_item_in_turn_id(&payload.turn_id, item);
     }
 
+    fn handle_apply_patch_approval_request(&mut self, payload: &ApplyPatchApprovalRequestEvent) {
+        let item = ThreadItem::FileChange {
+            id: payload.call_id.clone(),
+            changes: convert_patch_changes(&payload.changes),
+            status: PatchApplyStatus::InProgress,
+        };
+        if payload.turn_id.is_empty() {
+            self.upsert_item_in_current_turn(item);
+        } else {
+            self.upsert_item_in_turn_id(&payload.turn_id, item);
+        }
+    }
+
     fn handle_patch_apply_begin(&mut self, payload: &PatchApplyBeginEvent) {
         let item = ThreadItem::FileChange {
             id: payload.call_id.clone(),
@@ -1010,11 +1027,11 @@ fn format_file_change_diff(change: &codex_protocol::protocol::FileChange) -> Str
 }
 
 fn upsert_turn_item(items: &mut Vec<ThreadItem>, item: ThreadItem) {
-    if let Some(index) = items
-        .iter()
-        .rposition(|existing_item| existing_item.id() == item.id())
+    if let Some(existing_item) = items
+        .iter_mut()
+        .find(|existing_item| existing_item.id() == item.id())
     {
-        items[index] = item;
+        *existing_item = item;
         return;
     }
     items.push(item);
@@ -1080,6 +1097,7 @@ mod tests {
     use codex_protocol::protocol::AgentMessageEvent;
     use codex_protocol::protocol::AgentReasoningEvent;
     use codex_protocol::protocol::AgentReasoningRawContentEvent;
+    use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
     use codex_protocol::protocol::CodexErrorInfo;
     use codex_protocol::protocol::CompactedItem;
     use codex_protocol::protocol::DynamicToolCallResponseEvent;
@@ -1088,6 +1106,7 @@ mod tests {
     use codex_protocol::protocol::ItemStartedEvent;
     use codex_protocol::protocol::McpInvocation;
     use codex_protocol::protocol::McpToolCallEndEvent;
+    use codex_protocol::protocol::PatchApplyBeginEvent;
     use codex_protocol::protocol::ThreadRolledBackEvent;
     use codex_protocol::protocol::TurnAbortReason;
     use codex_protocol::protocol::TurnAbortedEvent;
@@ -1980,6 +1999,133 @@ mod tests {
         );
     }
 
+    #[test]
+    fn patch_apply_begin_updates_active_turn_snapshot_with_file_change() {
+        let turn_id = "turn-1";
+        let mut builder = ThreadHistoryBuilder::new();
+        let events = vec![
+            EventMsg::TurnStarted(TurnStartedEvent {
+                turn_id: turn_id.to_string(),
+                model_context_window: None,
+                collaboration_mode_kind: Default::default(),
+            }),
+            EventMsg::UserMessage(UserMessageEvent {
+                message: "apply patch".into(),
+                images: None,
+                text_elements: Vec::new(),
+                local_images: Vec::new(),
+            }),
+            EventMsg::PatchApplyBegin(PatchApplyBeginEvent {
+                call_id: "patch-call".into(),
+                turn_id: turn_id.to_string(),
+                auto_approved: false,
+                changes: [(
+                    PathBuf::from("README.md"),
+                    codex_protocol::protocol::FileChange::Add {
+                        content: "hello\n".into(),
+                    },
+                )]
+                .into_iter()
+                .collect(),
+            }),
+        ];
+
+        for event in &events {
+            builder.handle_event(event);
+        }
+
+        let snapshot = builder
+            .active_turn_snapshot()
+            .expect("active turn snapshot");
+        assert_eq!(snapshot.id, turn_id);
+        assert_eq!(snapshot.status, TurnStatus::InProgress);
+        assert_eq!(
+            snapshot.items,
+            vec![
+                ThreadItem::UserMessage {
+                    id: "item-1".into(),
+                    content: vec![UserInput::Text {
+                        text: "apply patch".into(),
+                        text_elements: Vec::new(),
+                    }],
+                },
+                ThreadItem::FileChange {
+                    id: "patch-call".into(),
+                    changes: vec![FileUpdateChange {
+                        path: "README.md".into(),
+                        kind: PatchChangeKind::Add,
+                        diff: "hello\n".into(),
+                    }],
+                    status: PatchApplyStatus::InProgress,
+                },
+            ]
+        );
+    }
+
+    #[test]
+    fn apply_patch_approval_request_updates_active_turn_snapshot_with_file_change() {
+        let turn_id = "turn-1";
+        let mut builder = ThreadHistoryBuilder::new();
+        let events = vec![
+            EventMsg::TurnStarted(TurnStartedEvent {
+                turn_id: turn_id.to_string(),
+                model_context_window: None,
+                collaboration_mode_kind: Default::default(),
+            }),
+            EventMsg::UserMessage(UserMessageEvent {
+                message: "apply patch".into(),
+                images: None,
+                text_elements: Vec::new(),
+                local_images: Vec::new(),
+            }),
+            EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent {
+                call_id: "patch-call".into(),
+                turn_id: turn_id.to_string(),
+                changes: [(
+                    PathBuf::from("README.md"),
+                    codex_protocol::protocol::FileChange::Add {
+                        content: "hello\n".into(),
+                    },
+                )]
+                .into_iter()
+                .collect(),
+                reason: None,
+                grant_root: None,
+            }),
+        ];
+
+        for event in &events {
+            builder.handle_event(event);
+        }
+
+        let snapshot = builder
+            .active_turn_snapshot()
+            .expect("active turn snapshot");
+        assert_eq!(snapshot.id, turn_id);
+        assert_eq!(snapshot.status, TurnStatus::InProgress);
+        assert_eq!(
+            snapshot.items,
+            vec![
+                ThreadItem::UserMessage {
+                    id: "item-1".into(),
+                    content: vec![UserInput::Text {
+                        text: "apply patch".into(),
+                        text_elements: Vec::new(),
+                    }],
+                },
+                ThreadItem::FileChange {
+                    id: "patch-call".into(),
+                    changes: vec![FileUpdateChange {
+                        path: "README.md".into(),
+                        kind: PatchChangeKind::Add,
+                        diff: "hello\n".into(),
+                    }],
+                    status: PatchApplyStatus::InProgress,
+                },
+            ]
+        );
+    }
+
     #[test]
     fn late_turn_complete_does_not_close_active_turn() {
         let events = vec![

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

@@ -5,6 +5,7 @@ use codex_protocol::ThreadId;
 use codex_protocol::config_types::ForcedLoginMethod;
 use codex_protocol::config_types::ReasoningSummary;
 use codex_protocol::config_types::SandboxMode;
+use codex_protocol::config_types::ServiceTier;
 use codex_protocol::config_types::Verbosity;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::openai_models::ReasoningEffort;
@@ -419,6 +420,13 @@ pub struct SendUserTurnParams {
     pub approval_policy: AskForApproval,
     pub sandbox_policy: SandboxPolicy,
     pub model: String,
+    #[serde(
+        default,
+        deserialize_with = "super::serde_helpers::deserialize_double_option",
+        serialize_with = "super::serde_helpers::serialize_double_option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub service_tier: Option<Option<ServiceTier>>,
     pub effort: Option<ReasoningEffort>,
     pub summary: ReasoningSummary,
     /// Optional JSON Schema used to constrain the final assistant message for this turn.
@@ -429,6 +437,55 @@ pub struct SendUserTurnParams {
 #[serde(rename_all = "camelCase")]
 pub struct SendUserTurnResponse {}
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+    use std::path::PathBuf;
+
+    #[test]
+    fn send_user_turn_params_preserve_explicit_null_service_tier() {
+        let params = SendUserTurnParams {
+            conversation_id: ThreadId::new(),
+            items: vec![],
+            cwd: PathBuf::from("/tmp"),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::DangerFullAccess,
+            model: "gpt-4.1".to_string(),
+            service_tier: Some(None),
+            effort: None,
+            summary: ReasoningSummary::Auto,
+            output_schema: None,
+        };
+
+        let serialized = serde_json::to_value(&params).expect("params should serialize");
+        assert_eq!(
+            serialized.get("serviceTier"),
+            Some(&serde_json::Value::Null)
+        );
+
+        let roundtrip: SendUserTurnParams =
+            serde_json::from_value(serialized).expect("params should deserialize");
+        assert_eq!(roundtrip.service_tier, Some(None));
+
+        let without_override = SendUserTurnParams {
+            conversation_id: ThreadId::new(),
+            items: vec![],
+            cwd: PathBuf::from("/tmp"),
+            approval_policy: AskForApproval::Never,
+            sandbox_policy: SandboxPolicy::DangerFullAccess,
+            model: "gpt-4.1".to_string(),
+            service_tier: None,
+            effort: None,
+            summary: ReasoningSummary::Auto,
+            output_schema: None,
+        };
+        let serialized_without_override =
+            serde_json::to_value(&without_override).expect("params should serialize");
+        assert_eq!(serialized_without_override.get("serviceTier"), None);
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 pub struct InterruptConversationParams {
@@ -555,6 +612,7 @@ pub struct LoginChatGptCompleteNotification {
 pub struct SessionConfiguredNotification {
     pub session_id: ThreadId,
     pub model: String,
+    pub service_tier: Option<ServiceTier>,
     pub reasoning_effort: Option<ReasoningEffort>,
     pub history_log_id: u64,
     #[ts(type = "number")]

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

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::path::PathBuf;
 
+use crate::RequestId;
 use crate::protocol::common::AuthMode;
 use codex_experimental_api_macros::ExperimentalApi;
 use codex_protocol::account::PlanType;
@@ -16,6 +17,7 @@ use codex_protocol::config_types::ModeKind;
 use codex_protocol::config_types::Personality;
 use codex_protocol::config_types::ReasoningSummary;
 use codex_protocol::config_types::SandboxMode as CoreSandboxMode;
+use codex_protocol::config_types::ServiceTier;
 use codex_protocol::config_types::Verbosity;
 use codex_protocol::config_types::WebSearchMode;
 use codex_protocol::items::AgentMessageContent as CoreAgentMessageContent;
@@ -31,6 +33,7 @@ use codex_protocol::models::MessagePhase;
 use codex_protocol::models::PermissionProfile as CorePermissionProfile;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::openai_models::InputModality;
+use codex_protocol::openai_models::ModelAvailabilityNux as CoreModelAvailabilityNux;
 use codex_protocol::openai_models::ReasoningEffort;
 use codex_protocol::openai_models::default_input_modalities;
 use codex_protocol::parse_command::ParsedCommand as CoreParsedCommand;
@@ -390,6 +393,7 @@ pub struct ProfileV2 {
     pub model: Option<String>,
     pub model_provider: Option<String>,
     pub approval_policy: Option<AskForApproval>,
+    pub service_tier: Option<ServiceTier>,
     pub model_reasoning_effort: Option<ReasoningEffort>,
     pub model_reasoning_summary: Option<ReasoningSummary>,
     pub model_verbosity: Option<Verbosity>,
@@ -501,6 +505,7 @@ pub struct Config {
     pub model_reasoning_effort: Option<ReasoningEffort>,
     pub model_reasoning_summary: Option<ReasoningSummary>,
     pub model_verbosity: Option<Verbosity>,
+    pub service_tier: Option<ServiceTier>,
     pub analytics: Option<AnalyticsConfig>,
     #[experimental("config/read.apps")]
     #[serde(default)]
@@ -811,8 +816,8 @@ impl From<CoreNetworkApprovalContext> for NetworkApprovalContext {
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
 pub struct AdditionalFileSystemPermissions {
-    pub read: Option<Vec<PathBuf>>,
-    pub write: Option<Vec<PathBuf>>,
+    pub read: Option<Vec<AbsolutePathBuf>>,
+    pub write: Option<Vec<AbsolutePathBuf>>,
 }
 
 impl From<CoreFileSystemPermissions> for AdditionalFileSystemPermissions {
@@ -1389,6 +1394,21 @@ pub struct ModelListParams {
     pub include_hidden: Option<bool>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ModelAvailabilityNux {
+    pub message: String,
+}
+
+impl From<CoreModelAvailabilityNux> for ModelAvailabilityNux {
+    fn from(value: CoreModelAvailabilityNux) -> Self {
+        Self {
+            message: value.message,
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -1396,6 +1416,8 @@ pub struct Model {
     pub id: String,
     pub model: String,
     pub upgrade: Option<String>,
+    pub upgrade_info: Option<ModelUpgradeInfo>,
+    pub availability_nux: Option<ModelAvailabilityNux>,
     pub display_name: String,
     pub description: String,
     pub hidden: bool,
@@ -1409,6 +1431,16 @@ pub struct Model {
     pub is_default: bool,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ModelUpgradeInfo {
+    pub model: String,
+    pub upgrade_copy: Option<String>,
+    pub model_link: Option<String>,
+    pub migration_markdown: Option<String>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -1759,6 +1791,14 @@ pub struct ThreadStartParams {
     pub model: Option<String>,
     #[ts(optional = nullable)]
     pub model_provider: Option<String>,
+    #[serde(
+        default,
+        deserialize_with = "super::serde_helpers::deserialize_double_option",
+        serialize_with = "super::serde_helpers::serialize_double_option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    #[ts(optional = nullable)]
+    pub service_tier: Option<Option<ServiceTier>>,
     #[ts(optional = nullable)]
     pub cwd: Option<String>,
     #[ts(optional = nullable)]
@@ -1821,6 +1861,7 @@ pub struct ThreadStartResponse {
     pub thread: Thread,
     pub model: String,
     pub model_provider: String,
+    pub service_tier: Option<ServiceTier>,
     pub cwd: PathBuf,
     pub approval_policy: AskForApproval,
     pub sandbox: SandboxPolicy,
@@ -1862,6 +1903,14 @@ pub struct ThreadResumeParams {
     pub model: Option<String>,
     #[ts(optional = nullable)]
     pub model_provider: Option<String>,
+    #[serde(
+        default,
+        deserialize_with = "super::serde_helpers::deserialize_double_option",
+        serialize_with = "super::serde_helpers::serialize_double_option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    #[ts(optional = nullable)]
+    pub service_tier: Option<Option<ServiceTier>>,
     #[ts(optional = nullable)]
     pub cwd: Option<String>,
     #[ts(optional = nullable)]
@@ -1890,6 +1939,7 @@ pub struct ThreadResumeResponse {
     pub thread: Thread,
     pub model: String,
     pub model_provider: String,
+    pub service_tier: Option<ServiceTier>,
     pub cwd: PathBuf,
     pub approval_policy: AskForApproval,
     pub sandbox: SandboxPolicy,
@@ -1922,6 +1972,14 @@ pub struct ThreadForkParams {
     pub model: Option<String>,
     #[ts(optional = nullable)]
     pub model_provider: Option<String>,
+    #[serde(
+        default,
+        deserialize_with = "super::serde_helpers::deserialize_double_option",
+        serialize_with = "super::serde_helpers::serialize_double_option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    #[ts(optional = nullable)]
+    pub service_tier: Option<Option<ServiceTier>>,
     #[ts(optional = nullable)]
     pub cwd: Option<String>,
     #[ts(optional = nullable)]
@@ -1948,6 +2006,7 @@ pub struct ThreadForkResponse {
     pub thread: Thread,
     pub model: String,
     pub model_provider: String,
+    pub service_tier: Option<ServiceTier>,
     pub cwd: PathBuf,
     pub approval_policy: AskForApproval,
     pub sandbox: SandboxPolicy,
@@ -2488,6 +2547,8 @@ pub struct Thread {
     pub id: String,
     /// Usually the first user message in the thread, if available.
     pub preview: String,
+    /// Whether the thread is ephemeral and should not be materialized on disk.
+    pub ephemeral: bool,
     /// Model provider used for this thread (for example, 'openai').
     pub model_provider: String,
     /// Unix timestamp (in seconds) when the thread was created.
@@ -2526,6 +2587,7 @@ pub struct Thread {
 #[ts(export_to = "v2/")]
 pub struct AccountUpdatedNotification {
     pub auth_mode: Option<AuthMode>,
+    pub plan_type: Option<PlanType>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -2805,6 +2867,15 @@ pub struct TurnStartParams {
     /// Override the model for this turn and subsequent turns.
     #[ts(optional = nullable)]
     pub model: Option<String>,
+    /// Override the service tier for this turn and subsequent turns.
+    #[serde(
+        default,
+        deserialize_with = "super::serde_helpers::deserialize_double_option",
+        serialize_with = "super::serde_helpers::serialize_double_option",
+        skip_serializing_if = "Option::is_none"
+    )]
+    #[ts(optional = nullable)]
+    pub service_tier: Option<Option<ServiceTier>>,
     /// Override the reasoning effort for this turn and subsequent turns.
     #[ts(optional = nullable)]
     pub effort: Option<ReasoningEffort>,
@@ -3717,6 +3788,14 @@ pub struct FileChangeOutputDeltaNotification {
     pub delta: String,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ServerRequestResolvedNotification {
+    pub thread_id: String,
+    pub request_id: RequestId,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
@@ -4142,6 +4221,37 @@ mod tests {
         AbsolutePathBuf::from_absolute_path(path).expect("path must be absolute")
     }
 
+    #[test]
+    fn command_execution_request_approval_rejects_relative_additional_permission_paths() {
+        let err = serde_json::from_value::<CommandExecutionRequestApprovalParams>(json!({
+            "threadId": "thr_123",
+            "turnId": "turn_123",
+            "itemId": "call_123",
+            "command": "cat file",
+            "cwd": "/tmp",
+            "commandActions": null,
+            "reason": null,
+            "networkApprovalContext": null,
+            "additionalPermissions": {
+                "network": null,
+                "fileSystem": {
+                    "read": ["relative/path"],
+                    "write": null
+                },
+                "macos": null
+            },
+            "proposedExecpolicyAmendment": null,
+            "proposedNetworkPolicyAmendments": null,
+            "availableDecisions": null
+        }))
+        .expect_err("relative additional permission paths should fail");
+        assert!(
+            err.to_string()
+                .contains("AbsolutePathBuf deserialized without a base path"),
+            "unexpected error: {err}"
+        );
+    }
+
     #[test]
     fn sandbox_policy_round_trips_external_sandbox_network_access() {
         let v2_policy = SandboxPolicy::ExternalSandbox {
@@ -4495,4 +4605,56 @@ mod tests {
             })
         );
     }
+
+    #[test]
+    fn thread_start_params_preserve_explicit_null_service_tier() {
+        let params: ThreadStartParams = serde_json::from_value(json!({ "serviceTier": null }))
+            .expect("params should deserialize");
+        assert_eq!(params.service_tier, Some(None));
+
+        let serialized = serde_json::to_value(&params).expect("params should serialize");
+        assert_eq!(
+            serialized.get("serviceTier"),
+            Some(&serde_json::Value::Null)
+        );
+
+        let serialized_without_override =
+            serde_json::to_value(ThreadStartParams::default()).expect("params should serialize");
+        assert_eq!(serialized_without_override.get("serviceTier"), None);
+    }
+
+    #[test]
+    fn turn_start_params_preserve_explicit_null_service_tier() {
+        let params: TurnStartParams = serde_json::from_value(json!({
+            "threadId": "thread_123",
+            "input": [],
+            "serviceTier": null
+        }))
+        .expect("params should deserialize");
+        assert_eq!(params.service_tier, Some(None));
+
+        let serialized = serde_json::to_value(&params).expect("params should serialize");
+        assert_eq!(
+            serialized.get("serviceTier"),
+            Some(&serde_json::Value::Null)
+        );
+
+        let without_override = TurnStartParams {
+            thread_id: "thread_123".to_string(),
+            input: vec![],
+            cwd: None,
+            approval_policy: None,
+            sandbox_policy: None,
+            model: None,
+            service_tier: None,
+            effort: None,
+            summary: None,
+            output_schema: None,
+            collaboration_mode: None,
+            personality: None,
+        };
+        let serialized_without_override =
+            serde_json::to_value(&without_override).expect("params should serialize");
+        assert_eq!(serialized_without_override.get("serviceTier"), None);
+    }
 }

codex-rs/app-server-test-client/README.md

@@ -18,6 +18,15 @@ cargo run -p codex-app-server-test-client -- \
 cargo run -p codex-app-server-test-client -- model-list
 ```
 
+## Watching Raw Inbound Traffic
+
+Initialize a connection, then print every inbound JSON-RPC message until you stop it with
+`Ctrl+C`:
+
+```bash
+cargo run -p codex-app-server-test-client -- watch
+```
+
 ## Testing Thread Rejoin Behavior
 
 Build and start an app server using commands above. The app-server log is written to `/tmp/codex-app-server-test-client/app-server.log`

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

@@ -15,6 +15,7 @@ use std::process::Command;
 use std::process::Stdio;
 use std::thread;
 use std::time::Duration;
+use std::time::SystemTime;
 
 use anyhow::Context;
 use anyhow::Result;
@@ -71,6 +72,7 @@ use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::W3cTraceContext;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use serde_json::Value;
@@ -104,6 +106,8 @@ const NOTIFICATIONS_TO_OPT_OUT: &[&str] = &[
     "item/reasoning/summaryTextDelta",
     "item/reasoning/textDelta",
 ];
+const APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
+const APP_SERVER_GRACEFUL_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(100);
 
 /// Minimal launcher that initializes the Codex app-server and logs the handshake.
 #[derive(Parser)]
@@ -188,6 +192,10 @@ enum CliCommand {
         /// Existing thread id to resume.
         thread_id: String,
     },
+    /// Initialize the app-server and dump all inbound messages until interrupted.
+    ///
+    /// This command does not auto-exit; stop it with SIGINT/SIGTERM/SIGKILL.
+    Watch,
     /// Start a V2 turn that elicits an ExecCommand approval.
     #[command(name = "trigger-cmd-approval")]
     TriggerCmdApproval {
@@ -291,6 +299,11 @@ pub fn run() -> Result<()> {
             let endpoint = resolve_endpoint(codex_bin, url)?;
             thread_resume_follow(&endpoint, &config_overrides, thread_id)
         }
+        CliCommand::Watch => {
+            ensure_dynamic_tools_unused(&dynamic_tools, "watch")?;
+            let endpoint = resolve_endpoint(codex_bin, url)?;
+            watch(&endpoint, &config_overrides)
+        }
         CliCommand::TriggerCmdApproval { user_message } => {
             let endpoint = resolve_endpoint(codex_bin, url)?;
             trigger_cmd_approval(&endpoint, &config_overrides, user_message, &dynamic_tools)
@@ -489,25 +502,26 @@ fn send_message(
     config_overrides: &[String],
     user_message: String,
 ) -> Result<()> {
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
+        let conversation = client.start_thread()?;
+        println!("< newConversation response: {conversation:?}");
 
-    let conversation = client.start_thread()?;
-    println!("< newConversation response: {conversation:?}");
+        let subscription = client.add_conversation_listener(&conversation.conversation_id)?;
+        println!("< addConversationListener response: {subscription:?}");
 
-    let subscription = client.add_conversation_listener(&conversation.conversation_id)?;
-    println!("< addConversationListener response: {subscription:?}");
+        let send_response =
+            client.send_user_message(&conversation.conversation_id, &user_message)?;
+        println!("< sendUserMessage response: {send_response:?}");
 
-    let send_response = client.send_user_message(&conversation.conversation_id, &user_message)?;
-    println!("< sendUserMessage response: {send_response:?}");
+        client.stream_conversation(&conversation.conversation_id)?;
 
-    client.stream_conversation(&conversation.conversation_id)?;
+        client.remove_thread_listener(subscription.subscription_id)?;
 
-    client.remove_thread_listener(subscription.subscription_id)?;
-
-    Ok(())
+        Ok(())
+    })
 }
 
 pub fn send_message_v2(
@@ -565,82 +579,85 @@ fn trigger_zsh_fork_multi_cmd_approval(
     let default_prompt = "Run this exact command using shell command execution without rewriting or splitting it: /usr/bin/true && /usr/bin/true";
     let message = user_message.unwrap_or_else(|| default_prompt.to_string());
 
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let thread_response = client.thread_start(ThreadStartParams {
-        dynamic_tools: dynamic_tools.clone(),
-        ..Default::default()
-    })?;
-    println!("< thread/start response: {thread_response:?}");
+        let thread_response = client.thread_start(ThreadStartParams {
+            dynamic_tools: dynamic_tools.clone(),
+            ..Default::default()
+        })?;
+        println!("< thread/start response: {thread_response:?}");
 
-    client.command_approval_behavior = match abort_on {
-        Some(index) => CommandApprovalBehavior::AbortOn(index),
-        None => CommandApprovalBehavior::AlwaysAccept,
-    };
-    client.command_approval_count = 0;
-    client.command_approval_item_ids.clear();
-    client.command_execution_statuses.clear();
-    client.last_turn_status = None;
+        client.command_approval_behavior = match abort_on {
+            Some(index) => CommandApprovalBehavior::AbortOn(index),
+            None => CommandApprovalBehavior::AlwaysAccept,
+        };
+        client.command_approval_count = 0;
+        client.command_approval_item_ids.clear();
+        client.command_execution_statuses.clear();
+        client.last_turn_status = None;
 
-    let mut turn_params = TurnStartParams {
-        thread_id: thread_response.thread.id.clone(),
-        input: vec![V2UserInput::Text {
-            text: message,
-            text_elements: Vec::new(),
-        }],
-        ..Default::default()
-    };
-    turn_params.approval_policy = Some(AskForApproval::OnRequest);
-    turn_params.sandbox_policy = Some(SandboxPolicy::ReadOnly {
-        access: ReadOnlyAccess::FullAccess,
-    });
+        let mut turn_params = TurnStartParams {
+            thread_id: thread_response.thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: message,
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        };
+        turn_params.approval_policy = Some(AskForApproval::OnRequest);
+        turn_params.sandbox_policy = Some(SandboxPolicy::ReadOnly {
+            access: ReadOnlyAccess::FullAccess,
+        });
 
-    let turn_response = client.turn_start(turn_params)?;
-    println!("< turn/start response: {turn_response:?}");
-    client.stream_turn(&thread_response.thread.id, &turn_response.turn.id)?;
+        let turn_response = client.turn_start(turn_params)?;
+        println!("< turn/start response: {turn_response:?}");
+        client.stream_turn(&thread_response.thread.id, &turn_response.turn.id)?;
 
-    if client.command_approval_count < min_approvals {
-        bail!(
-            "expected at least {min_approvals} command approvals, got {}",
-            client.command_approval_count
-        );
-    }
-    let mut approvals_per_item = std::collections::BTreeMap::new();
-    for item_id in &client.command_approval_item_ids {
-        *approvals_per_item.entry(item_id.clone()).or_insert(0usize) += 1;
-    }
-    let max_approvals_for_one_item = approvals_per_item.values().copied().max().unwrap_or(0);
-    if max_approvals_for_one_item < min_approvals {
-        bail!(
-            "expected at least {min_approvals} approvals for one command item, got max {max_approvals_for_one_item} with map {approvals_per_item:?}"
-        );
-    }
-
-    let last_command_status = client.command_execution_statuses.last();
-    if abort_on.is_none() {
-        if last_command_status != Some(&CommandExecutionStatus::Completed) {
-            bail!("expected completed command execution, got {last_command_status:?}");
-        }
-        if client.last_turn_status != Some(TurnStatus::Completed) {
+        if client.command_approval_count < min_approvals {
             bail!(
-                "expected completed turn in all-accept flow, got {:?}",
-                client.last_turn_status
+                "expected at least {min_approvals} command approvals, got {}",
+                client.command_approval_count
             );
         }
-    } else if last_command_status == Some(&CommandExecutionStatus::Completed) {
-        bail!(
-            "expected non-completed command execution in mixed approval/decline flow, got {last_command_status:?}"
+        let mut approvals_per_item = std::collections::BTreeMap::new();
+        for item_id in &client.command_approval_item_ids {
+            *approvals_per_item.entry(item_id.clone()).or_insert(0usize) += 1;
+        }
+        let max_approvals_for_one_item = approvals_per_item.values().copied().max().unwrap_or(0);
+        if max_approvals_for_one_item < min_approvals {
+            bail!(
+                "expected at least {min_approvals} approvals for one command item, got max {max_approvals_for_one_item} with map {approvals_per_item:?}"
+            );
+        }
+
+        let last_command_status = client.command_execution_statuses.last();
+        if abort_on.is_none() {
+            if last_command_status != Some(&CommandExecutionStatus::Completed) {
+                bail!("expected completed command execution, got {last_command_status:?}");
+            }
+            if client.last_turn_status != Some(TurnStatus::Completed) {
+                bail!(
+                    "expected completed turn in all-accept flow, got {:?}",
+                    client.last_turn_status
+                );
+            }
+        } else if last_command_status == Some(&CommandExecutionStatus::Completed) {
+            bail!(
+                "expected non-completed command execution in mixed approval/decline flow, got {last_command_status:?}"
+            );
+        }
+
+        println!(
+            "[zsh-fork multi-approval summary] approvals={}, approvals_per_item={approvals_per_item:?}, command_statuses={:?}, turn_status={:?}",
+            client.command_approval_count,
+            client.command_execution_statuses,
+            client.last_turn_status
         );
-    }
 
-    println!(
-        "[zsh-fork multi-approval summary] approvals={}, approvals_per_item={approvals_per_item:?}, command_statuses={:?}, turn_status={:?}",
-        client.command_approval_count, client.command_execution_statuses, client.last_turn_status
-    );
-
-    Ok(())
+        Ok(())
+    })
 }
 
 fn resume_message_v2(
@@ -652,30 +669,30 @@ fn resume_message_v2(
 ) -> Result<()> {
     ensure_dynamic_tools_unused(dynamic_tools, "resume-message-v2")?;
 
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
+        let resume_response = client.thread_resume(ThreadResumeParams {
+            thread_id,
+            ..Default::default()
+        })?;
+        println!("< thread/resume response: {resume_response:?}");
 
-    let resume_response = client.thread_resume(ThreadResumeParams {
-        thread_id,
-        ..Default::default()
-    })?;
-    println!("< thread/resume response: {resume_response:?}");
+        let turn_response = client.turn_start(TurnStartParams {
+            thread_id: resume_response.thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: user_message,
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })?;
+        println!("< turn/start response: {turn_response:?}");
 
-    let turn_response = client.turn_start(TurnStartParams {
-        thread_id: resume_response.thread.id.clone(),
-        input: vec![V2UserInput::Text {
-            text: user_message,
-            text_elements: Vec::new(),
-        }],
-        ..Default::default()
-    })?;
-    println!("< turn/start response: {turn_response:?}");
+        client.stream_turn(&resume_response.thread.id, &turn_response.turn.id)?;
 
-    client.stream_turn(&resume_response.thread.id, &turn_response.turn.id)?;
-
-    Ok(())
+        Ok(())
+    })
 }
 
 fn thread_resume_follow(
@@ -698,6 +715,16 @@ fn thread_resume_follow(
     client.stream_notifications_forever()
 }
 
+fn watch(endpoint: &Endpoint, config_overrides: &[String]) -> Result<()> {
+    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+
+    let initialize = client.initialize()?;
+    println!("< initialize response: {initialize:?}");
+    println!("< streaming inbound messages until process is terminated");
+
+    client.stream_notifications_forever()
+}
+
 fn trigger_cmd_approval(
     endpoint: &Endpoint,
     config_overrides: &[String],
@@ -768,34 +795,34 @@ fn send_message_v2_with_policies(
     sandbox_policy: Option<SandboxPolicy>,
     dynamic_tools: &Option<Vec<DynamicToolSpec>>,
 ) -> Result<()> {
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize_with_experimental_api(experimental_api)?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize_with_experimental_api(experimental_api)?;
-    println!("< initialize response: {initialize:?}");
+        let thread_response = client.thread_start(ThreadStartParams {
+            dynamic_tools: dynamic_tools.clone(),
+            ..Default::default()
+        })?;
+        println!("< thread/start response: {thread_response:?}");
+        let mut turn_params = TurnStartParams {
+            thread_id: thread_response.thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: user_message,
+                // Test client sends plain text without UI element ranges.
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        };
+        turn_params.approval_policy = approval_policy;
+        turn_params.sandbox_policy = sandbox_policy;
 
-    let thread_response = client.thread_start(ThreadStartParams {
-        dynamic_tools: dynamic_tools.clone(),
-        ..Default::default()
-    })?;
-    println!("< thread/start response: {thread_response:?}");
-    let mut turn_params = TurnStartParams {
-        thread_id: thread_response.thread.id.clone(),
-        input: vec![V2UserInput::Text {
-            text: user_message,
-            // Test client sends plain text without UI element ranges.
-            text_elements: Vec::new(),
-        }],
-        ..Default::default()
-    };
-    turn_params.approval_policy = approval_policy;
-    turn_params.sandbox_policy = sandbox_policy;
+        let turn_response = client.turn_start(turn_params)?;
+        println!("< turn/start response: {turn_response:?}");
 
-    let turn_response = client.turn_start(turn_params)?;
-    println!("< turn/start response: {turn_response:?}");
+        client.stream_turn(&thread_response.thread.id, &turn_response.turn.id)?;
 
-    client.stream_turn(&thread_response.thread.id, &turn_response.turn.id)?;
-
-    Ok(())
+        Ok(())
+    })
 }
 
 fn send_follow_up_v2(
@@ -805,119 +832,130 @@ fn send_follow_up_v2(
     follow_up_message: String,
     dynamic_tools: &Option<Vec<DynamicToolSpec>>,
 ) -> Result<()> {
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
+        let thread_response = client.thread_start(ThreadStartParams {
+            dynamic_tools: dynamic_tools.clone(),
+            ..Default::default()
+        })?;
+        println!("< thread/start response: {thread_response:?}");
 
-    let thread_response = client.thread_start(ThreadStartParams {
-        dynamic_tools: dynamic_tools.clone(),
-        ..Default::default()
-    })?;
-    println!("< thread/start response: {thread_response:?}");
+        let first_turn_params = TurnStartParams {
+            thread_id: thread_response.thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: first_message,
+                // Test client sends plain text without UI element ranges.
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        };
+        let first_turn_response = client.turn_start(first_turn_params)?;
+        println!("< turn/start response (initial): {first_turn_response:?}");
+        client.stream_turn(&thread_response.thread.id, &first_turn_response.turn.id)?;
 
-    let first_turn_params = TurnStartParams {
-        thread_id: thread_response.thread.id.clone(),
-        input: vec![V2UserInput::Text {
-            text: first_message,
-            // Test client sends plain text without UI element ranges.
-            text_elements: Vec::new(),
-        }],
-        ..Default::default()
-    };
-    let first_turn_response = client.turn_start(first_turn_params)?;
-    println!("< turn/start response (initial): {first_turn_response:?}");
-    client.stream_turn(&thread_response.thread.id, &first_turn_response.turn.id)?;
+        let follow_up_params = TurnStartParams {
+            thread_id: thread_response.thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: follow_up_message,
+                // Test client sends plain text without UI element ranges.
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        };
+        let follow_up_response = client.turn_start(follow_up_params)?;
+        println!("< turn/start response (follow-up): {follow_up_response:?}");
+        client.stream_turn(&thread_response.thread.id, &follow_up_response.turn.id)?;
 
-    let follow_up_params = TurnStartParams {
-        thread_id: thread_response.thread.id.clone(),
-        input: vec![V2UserInput::Text {
-            text: follow_up_message,
-            // Test client sends plain text without UI element ranges.
-            text_elements: Vec::new(),
-        }],
-        ..Default::default()
-    };
-    let follow_up_response = client.turn_start(follow_up_params)?;
-    println!("< turn/start response (follow-up): {follow_up_response:?}");
-    client.stream_turn(&thread_response.thread.id, &follow_up_response.turn.id)?;
-
-    Ok(())
+        Ok(())
+    })
 }
 
 fn test_login(endpoint: &Endpoint, config_overrides: &[String]) -> Result<()> {
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
-
-    let login_response = client.login_chat_gpt()?;
-    println!("< loginChatGpt response: {login_response:?}");
-    println!(
-        "Open the following URL in your browser to continue:\n{}",
-        login_response.auth_url
-    );
-
-    let completion = client.wait_for_login_completion(&login_response.login_id)?;
-    println!("< loginChatGptComplete notification: {completion:?}");
-
-    if completion.success {
-        println!("Login succeeded.");
-        Ok(())
-    } else {
-        bail!(
-            "login failed: {}",
-            completion
-                .error
-                .as_deref()
-                .unwrap_or("unknown error from loginChatGptComplete")
+        let login_response = client.login_chat_gpt()?;
+        println!("< loginChatGpt response: {login_response:?}");
+        println!(
+            "Open the following URL in your browser to continue:\n{}",
+            login_response.auth_url
         );
-    }
+
+        let completion = client.wait_for_login_completion(&login_response.login_id)?;
+        println!("< loginChatGptComplete notification: {completion:?}");
+
+        if completion.success {
+            println!("Login succeeded.");
+            Ok(())
+        } else {
+            bail!(
+                "login failed: {}",
+                completion
+                    .error
+                    .as_deref()
+                    .unwrap_or("unknown error from loginChatGptComplete")
+            );
+        }
+    })
 }
 
 fn get_account_rate_limits(endpoint: &Endpoint, config_overrides: &[String]) -> Result<()> {
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
+        let response = client.get_account_rate_limits()?;
+        println!("< account/rateLimits/read response: {response:?}");
 
-    let response = client.get_account_rate_limits()?;
-    println!("< account/rateLimits/read response: {response:?}");
-
-    Ok(())
+        Ok(())
+    })
 }
 
 fn model_list(endpoint: &Endpoint, config_overrides: &[String]) -> Result<()> {
-    let mut client = CodexClient::connect(endpoint, config_overrides)?;
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
 
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
+        let response = client.model_list(ModelListParams::default())?;
+        println!("< model/list response: {response:?}");
 
-    let response = client.model_list(ModelListParams::default())?;
-    println!("< model/list response: {response:?}");
-
-    Ok(())
+        Ok(())
+    })
 }
 
 fn thread_list(endpoint: &Endpoint, config_overrides: &[String], limit: u32) -> Result<()> {
+    with_client(endpoint, config_overrides, |client| {
+        let initialize = client.initialize()?;
+        println!("< initialize response: {initialize:?}");
+
+        let response = client.thread_list(ThreadListParams {
+            cursor: None,
+            limit: Some(limit),
+            sort_key: None,
+            model_providers: None,
+            source_kinds: None,
+            archived: None,
+            cwd: None,
+            search_term: None,
+        })?;
+        println!("< thread/list response: {response:?}");
+
+        Ok(())
+    })
+}
+
+fn with_client<T>(
+    endpoint: &Endpoint,
+    config_overrides: &[String],
+    f: impl FnOnce(&mut CodexClient) -> Result<T>,
+) -> Result<T> {
     let mut client = CodexClient::connect(endpoint, config_overrides)?;
-
-    let initialize = client.initialize()?;
-    println!("< initialize response: {initialize:?}");
-
-    let response = client.thread_list(ThreadListParams {
-        cursor: None,
-        limit: Some(limit),
-        sort_key: None,
-        model_providers: None,
-        source_kinds: None,
-        archived: None,
-        cwd: None,
-        search_term: None,
-    })?;
-    println!("< thread/list response: {response:?}");
-
-    Ok(())
+    let result = f(&mut client);
+    client.print_trace_summary();
+    result
 }
 
 fn ensure_dynamic_tools_unused(
@@ -974,6 +1012,8 @@ struct CodexClient {
     command_approval_item_ids: Vec<String>,
     command_execution_statuses: Vec<CommandExecutionStatus>,
     last_turn_status: Option<TurnStatus>,
+    trace_id: String,
+    trace_root_span_id: String,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -1033,6 +1073,8 @@ impl CodexClient {
             command_approval_item_ids: Vec::new(),
             command_execution_statuses: Vec::new(),
             last_turn_status: None,
+            trace_id: generate_trace_id(),
+            trace_root_span_id: generate_parent_span_id(),
         })
     }
 
@@ -1054,6 +1096,8 @@ impl CodexClient {
             command_approval_item_ids: Vec::new(),
             command_execution_statuses: Vec::new(),
             last_turn_status: None,
+            trace_id: generate_trace_id(),
+            trace_root_span_id: generate_parent_span_id(),
         })
     }
 
@@ -1419,12 +1463,32 @@ impl CodexClient {
     }
 
     fn write_request(&mut self, request: &ClientRequest) -> Result<()> {
-        let request_json = serde_json::to_string(request)?;
-        let request_pretty = serde_json::to_string_pretty(request)?;
+        let request = self.jsonrpc_request_with_trace(request)?;
+        let request_json = serde_json::to_string(&request)?;
+        let request_pretty = serde_json::to_string_pretty(&request)?;
         print_multiline_with_prefix("> ", &request_pretty);
         self.write_payload(&request_json)
     }
 
+    fn jsonrpc_request_with_trace(&self, request: &ClientRequest) -> Result<JSONRPCRequest> {
+        let request_value = serde_json::to_value(request)?;
+        let mut request: JSONRPCRequest = serde_json::from_value(request_value)
+            .context("client request was not a valid JSON-RPC request")?;
+        request.trace = Some(W3cTraceContext {
+            traceparent: Some(format!(
+                "00-{}-{}-01",
+                self.trace_id, self.trace_root_span_id
+            )),
+            tracestate: None,
+        });
+        Ok(request)
+    }
+
+    fn print_trace_summary(&self) {
+        println!("\n[Datadog trace]");
+        println!("go/trace/{}\n", self.trace_id);
+    }
+
     fn wait_for_response<T>(&mut self, request_id: RequestId, method: &str) -> Result<T>
     where
         T: DeserializeOwned,
@@ -1690,6 +1754,15 @@ impl CodexClient {
     }
 }
 
+fn generate_trace_id() -> String {
+    Uuid::new_v4().simple().to_string()
+}
+
+fn generate_parent_span_id() -> String {
+    let uuid = Uuid::new_v4().simple().to_string();
+    uuid[..16].to_string()
+}
+
 fn print_multiline_with_prefix(prefix: &str, payload: &str) {
     for line in payload.lines() {
         println!("{prefix}{line}");
@@ -1709,11 +1782,18 @@ impl Drop for CodexClient {
             return;
         }
 
-        thread::sleep(Duration::from_millis(100));
+        let deadline = SystemTime::now() + APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT;
+        loop {
+            if let Ok(Some(status)) = child.try_wait() {
+                println!("[codex app-server exited: {status}]");
+                return;
+            }
 
-        if let Ok(Some(status)) = child.try_wait() {
-            println!("[codex app-server exited: {status}]");
-            return;
+            if SystemTime::now() >= deadline {
+                break;
+            }
+
+            thread::sleep(APP_SERVER_GRACEFUL_SHUTDOWN_POLL_INTERVAL);
         }
 
         let _ = child.kill();

codex-rs/app-server/Cargo.toml

@@ -21,6 +21,7 @@ async-trait = { workspace = true }
 codex-arg0 = { workspace = true }
 codex-cloud-requirements = { workspace = true }
 codex-core = { workspace = true }
+codex-otel = { workspace = true }
 codex-shell-command = { workspace = true }
 codex-utils-cli = { workspace = true }
 codex-backend-client = { workspace = true }

codex-rs/app-server/README.md

@@ -62,7 +62,8 @@ Use the thread APIs to create, list, or archive conversations. Drive a conversat
 
 - Initialize once per connection: Immediately after opening a transport connection, send an `initialize` request with your client metadata, then emit an `initialized` notification. Any other request on that connection before this handshake gets rejected.
 - Start (or resume) a thread: Call `thread/start` to open a fresh conversation. The response returns the thread object and you’ll also get a `thread/started` notification. If you’re continuing an existing conversation, call `thread/resume` with its ID instead. If you want to branch from an existing conversation, call `thread/fork` to create a new thread id with copied history.
-- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, etc. This immediately returns the new turn object and triggers a `turn/started` notification.
+  The returned `thread.ephemeral` flag tells you whether the session is intentionally in-memory only; when it is `true`, `thread.path` is `null`.
+- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, etc. This immediately returns the new turn object. The app-server emits `turn/started` when that turn actually begins running.
 - Stream events: After `turn/start`, keep reading JSON-RPC notifications on stdout. You’ll see `item/started`, `item/completed`, deltas like `item/agentMessage/delta`, tool progress, etc. These represent streaming model output plus any side effects (commands, tool calls, reasoning notes).
 - Finish the turn: When the model is done (or the turn is interrupted via making the `turn/interrupt` call), the server sends `turn/completed` with the final turn state and token usage.
 
@@ -119,16 +120,16 @@ Example with notification opt-out:
 
 ## API Overview
 
-- `thread/start` — create a new thread; emits `thread/started` and auto-subscribes you to turn/item events for that thread.
+- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread.
 - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it.
-- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; emits `thread/started` and auto-subscribes you to turn/item events for the new thread.
+- `thread/fork` — fork an existing thread into a new thread id by copying the stored history; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for the new thread.
 - `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
 - `thread/loaded/list` — list the thread ids currently loaded in memory.
 - `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
 - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
 - `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
 - `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
-- `thread/name/set` — set or update a thread’s user-facing name; returns `{}` on success. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
+- `thread/name/set` — set or update a thread’s user-facing name for either a loaded thread or a persisted rollout; returns `{}` on success. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
 - `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
 - `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
 - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
@@ -142,7 +143,7 @@ Example with notification opt-out:
 - `thread/realtime/stop` — stop the active realtime session for the thread (experimental); returns `{}`.
 - `review/start` — kick off Codex’s automated reviewer for a thread; responds like `turn/start` and emits `item/started`/`item/completed` notifications with `enteredReviewMode` and `exitedReviewMode` items, plus a final assistant `agentMessage` containing the review.
 - `command/exec` — run a single command under the server sandbox without starting a thread/turn (handy for utilities and validation).
-- `model/list` — list available models (set `includeHidden: true` to include entries with `hidden: true`), with reasoning effort options and optional `upgrade` model ids.
+- `model/list` — list available models (set `includeHidden: true` to include entries with `hidden: true`), with reasoning effort options, optional legacy `upgrade` model ids, optional `upgradeInfo` metadata (`model`, `upgradeCopy`, `modelLink`, `migrationMarkdown`), and optional `availabilityNux` metadata.
 - `experimentalFeature/list` — list feature flags with stage metadata (`beta`, `underDevelopment`, `stable`, etc.), enabled/default-enabled state, and cursor pagination. For non-beta flags, `displayName`/`description`/`announcement` are `null`.
 - `collaborationMode/list` — list available collaboration mode presets (experimental, no pagination). This response omits built-in developer instructions; clients should either pass `settings.developer_instructions: null` when setting a mode to use Codex's built-in instructions, or provide their own instructions explicitly.
 - `skills/list` — list skills for one or more `cwd` values (optional `forceReload`).
@@ -272,10 +273,11 @@ When `nextCursor` is `null`, you’ve reached the final page.
 
 ### Example: Track thread status changes
 
-`thread/status/changed` is emitted whenever a loaded thread's status changes:
+`thread/status/changed` is emitted whenever a loaded thread's status changes after it has already been introduced to the client:
 
 - Includes `threadId` and the new `status`.
 - Status can be `notLoaded`, `idle`, `systemError`, or `active` (with `activeFlags`; `active` implies running).
+- `thread/start`, `thread/fork`, and detached review threads do not emit a separate initial `thread/status/changed`; their `thread/started` notification already carries the current `thread.status`.
 
 ```json
 { "method": "thread/status/changed", "params": {
@@ -619,7 +621,7 @@ Because audio is intentionally separate from `ThreadItem`, clients can opt out o
 
 ### Turn events
 
-The app-server streams JSON-RPC notifications while a turn is running. Each turn starts with `turn/started` (initial `turn`) and ends with `turn/completed` (final `turn` status). Token usage events stream separately via `thread/tokenUsage/updated`. Clients subscribe to the events they care about, rendering each item incrementally as updates arrive. The per-item lifecycle is always: `item/started` → zero or more item-specific deltas → `item/completed`.
+The app-server streams JSON-RPC notifications while a turn is running. Each turn emits `turn/started` when it begins running and ends with `turn/completed` (final `turn` status). Token usage events stream separately via `thread/tokenUsage/updated`. Clients subscribe to the events they care about, rendering each item incrementally as updates arrive. The per-item lifecycle is always: `item/started` → zero or more item-specific deltas → `item/completed`.
 
 - `turn/started` — `{ turn }` with the turn id, empty `items`, and `status: "inProgress"`.
 - `turn/completed` — `{ turn }` where `turn.status` is `completed`, `interrupted`, or `failed`; failures carry `{ error: { message, codexErrorInfo?, additionalDetails? } }`.
@@ -710,9 +712,10 @@ Certain actions (shell commands or modifying files) may require explicit user ap
 Order of messages:
 
 1. `item/started` — shows the pending `commandExecution` item with `command`, `cwd`, and other fields so you can render the proposed action.
-2. `item/commandExecution/requestApproval` (request) — carries the same `itemId`, `threadId`, `turnId`, optionally `approvalId` (for subcommand callbacks), and `reason`. For normal command approvals, it also includes `command`, `cwd`, and `commandActions` for friendly display. When `initialize.params.capabilities.experimentalApi = true`, it may also include experimental `additionalPermissions` describing requested per-command sandbox access. For network-only approvals, those command fields may be omitted and `networkApprovalContext` is provided instead. Optional persistence hints may also be included via `proposedExecpolicyAmendment` and `proposedNetworkPolicyAmendments`. Clients can prefer `availableDecisions` when present to render the exact set of choices the server wants to expose, while still falling back to the older heuristics if it is omitted.
+2. `item/commandExecution/requestApproval` (request) — carries the same `itemId`, `threadId`, `turnId`, optionally `approvalId` (for subcommand callbacks), and `reason`. For normal command approvals, it also includes `command`, `cwd`, and `commandActions` for friendly display. When `initialize.params.capabilities.experimentalApi = true`, it may also include experimental `additionalPermissions` describing requested per-command sandbox access; any filesystem paths in that payload are absolute on the wire. For network-only approvals, those command fields may be omitted and `networkApprovalContext` is provided instead. Optional persistence hints may also be included via `proposedExecpolicyAmendment` and `proposedNetworkPolicyAmendments`. Clients can prefer `availableDecisions` when present to render the exact set of choices the server wants to expose, while still falling back to the older heuristics if it is omitted.
 3. Client response — for example `{ "decision": "accept" }`, `{ "decision": "acceptForSession" }`, `{ "decision": { "acceptWithExecpolicyAmendment": { "execpolicy_amendment": [...] } } }`, `{ "decision": { "applyNetworkPolicyAmendment": { "network_policy_amendment": { "host": "example.com", "action": "allow" } } } }`, `{ "decision": "decline" }`, or `{ "decision": "cancel" }`.
-4. `item/completed` — final `commandExecution` item with `status: "completed" | "failed" | "declined"` and execution output. Render this as the authoritative result.
+4. `serverRequest/resolved` — `{ threadId, requestId }` confirms the pending request has been resolved or cleared, including lifecycle cleanup on turn start/complete/interrupt.
+5. `item/completed` — final `commandExecution` item with `status: "completed" | "failed" | "declined"` and execution output. Render this as the authoritative result.
 
 ### File change approvals
 
@@ -721,10 +724,15 @@ Order of messages:
 1. `item/started` — emits a `fileChange` item with `changes` (diff chunk summaries) and `status: "inProgress"`. Show the proposed edits and paths to the user.
 2. `item/fileChange/requestApproval` (request) — includes `itemId`, `threadId`, `turnId`, and an optional `reason`.
 3. Client response — `{ "decision": "accept" }` or `{ "decision": "decline" }`.
-4. `item/completed` — returns the same `fileChange` item with `status` updated to `completed`, `failed`, or `declined` after the patch attempt. Rely on this to show success/failure and finalize the diff state in your UI.
+4. `serverRequest/resolved` — `{ threadId, requestId }` confirms the pending request has been resolved or cleared, including lifecycle cleanup on turn start/complete/interrupt.
+5. `item/completed` — returns the same `fileChange` item with `status` updated to `completed`, `failed`, or `declined` after the patch attempt. Rely on this to show success/failure and finalize the diff state in your UI.
 
 UI guidance for IDEs: surface an approval dialog as soon as the request arrives. The turn will proceed after the server receives a response to the approval request. The terminal `item/completed` notification will be sent with the appropriate status.
 
+### request_user_input
+
+When the client responds to `item/tool/requestUserInput`, the server emits `serverRequest/resolved` with `{ threadId, requestId }`. If the pending request is cleared by turn start, turn completion, or turn interruption before the client answers, the server emits the same notification for that cleanup.
+
 ### Dynamic tool calls (experimental)
 
 `dynamicTools` on `thread/start` and the corresponding `item/tool/call` request/response flow are experimental APIs. To enable them, set `initialize.params.capabilities.experimentalApi = true`.
@@ -945,7 +953,7 @@ The JSON-RPC auth/account surface exposes request/response methods plus server-i
 
 ### Authentication modes
 
-Codex supports these authentication modes. The current mode is surfaced in `account/updated` (`authMode`) and can be inferred from `account/read`.
+Codex supports these authentication modes. The current mode is surfaced in `account/updated` (`authMode`), which also includes the current ChatGPT `planType` when available, and can be inferred from `account/read`.
 
 - **API key (`apiKey`)**: Caller supplies an OpenAI API key via `account/login/start` with `type: "apiKey"`. The API key is saved and used for API requests.
 - **ChatGPT managed (`chatgpt`)** (recommended): Codex owns the ChatGPT OAuth flow and refresh tokens. Start via `account/login/start` with `type: "chatgpt"`; Codex persists tokens to disk and refreshes them automatically.
@@ -957,7 +965,7 @@ Codex supports these authentication modes. The current mode is surfaced in `acco
 - `account/login/completed` (notify) — emitted when a login attempt finishes (success or error).
 - `account/login/cancel` — cancel a pending ChatGPT login by `loginId`.
 - `account/logout` — sign out; triggers `account/updated`.
-- `account/updated` (notify) — emitted whenever auth mode changes (`authMode`: `apikey`, `chatgpt`, or `null`).
+- `account/updated` (notify) — emitted whenever auth mode changes (`authMode`: `apikey`, `chatgpt`, or `null`) and includes the current ChatGPT `planType` when available.
 - `account/rateLimits/read` — fetch ChatGPT rate limits; updates arrive via `account/rateLimits/updated` (notify).
 - `account/rateLimits/updated` (notify) — emitted whenever a user's ChatGPT rate limits change.
 - `mcpServer/oauthLogin/completed` (notify) — emitted after a `mcpServer/oauth/login` flow finishes for a server; payload includes `{ name, success, error? }`.
@@ -1001,7 +1009,7 @@ Field notes:
 3. Notifications:
    ```json
    { "method": "account/login/completed", "params": { "loginId": null, "success": true, "error": null } }
-   { "method": "account/updated", "params": { "authMode": "apikey" } }
+   { "method": "account/updated", "params": { "authMode": "apikey", "planType": null } }
    ```
 
 ### 3) Log in with ChatGPT (browser flow)
@@ -1015,7 +1023,7 @@ Field notes:
 3. Wait for notifications:
    ```json
    { "method": "account/login/completed", "params": { "loginId": "<uuid>", "success": true, "error": null } }
-   { "method": "account/updated", "params": { "authMode": "chatgpt" } }
+   { "method": "account/updated", "params": { "authMode": "chatgpt", "planType": "plus" } }
    ```
 
 ### 4) Cancel a ChatGPT login
@@ -1030,7 +1038,7 @@ Field notes:
 ```json
 { "method": "account/logout", "id": 5 }
 { "id": 5, "result": {} }
-{ "method": "account/updated", "params": { "authMode": null } }
+{ "method": "account/updated", "params": { "authMode": null, "planType": null } }
 ```
 
 ### 6) Rate limits (ChatGPT)

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

@@ -0,0 +1,101 @@
+use crate::message_processor::ConnectionSessionState;
+use crate::outgoing_message::ConnectionId;
+use crate::transport::AppServerTransport;
+use codex_app_server_protocol::InitializeParams;
+use codex_app_server_protocol::JSONRPCRequest;
+use codex_otel::set_parent_from_context;
+use codex_otel::set_parent_from_w3c_trace_context;
+use codex_otel::traceparent_context_from_env;
+use codex_protocol::protocol::W3cTraceContext;
+use tracing::Span;
+use tracing::field;
+use tracing::info_span;
+
+pub(crate) fn request_span(
+    request: &JSONRPCRequest,
+    transport: AppServerTransport,
+    connection_id: ConnectionId,
+    session: &ConnectionSessionState,
+) -> Span {
+    let span = info_span!(
+        "app_server.request",
+        otel.kind = "server",
+        otel.name = request.method.as_str(),
+        rpc.system = "jsonrpc",
+        rpc.method = request.method.as_str(),
+        rpc.transport = transport_name(transport),
+        rpc.request_id = ?request.id,
+        app_server.connection_id = ?connection_id,
+        app_server.api_version = "v2",
+        app_server.client_name = field::Empty,
+        app_server.client_version = field::Empty,
+    );
+
+    let initialize_client_info = initialize_client_info(request);
+    if let Some(client_name) = client_name(initialize_client_info.as_ref(), session) {
+        span.record("app_server.client_name", client_name);
+    }
+    if let Some(client_version) = client_version(initialize_client_info.as_ref(), session) {
+        span.record("app_server.client_version", client_version);
+    }
+
+    if let Some(traceparent) = request
+        .trace
+        .as_ref()
+        .and_then(|trace| trace.traceparent.as_deref())
+    {
+        let trace = W3cTraceContext {
+            traceparent: Some(traceparent.to_string()),
+            tracestate: request
+                .trace
+                .as_ref()
+                .and_then(|value| value.tracestate.clone()),
+        };
+        if !set_parent_from_w3c_trace_context(&span, &trace) {
+            tracing::warn!(
+                rpc_method = request.method.as_str(),
+                rpc_request_id = ?request.id,
+                "ignoring invalid inbound request trace carrier"
+            );
+        }
+    } else if let Some(context) = traceparent_context_from_env() {
+        set_parent_from_context(&span, context);
+    }
+
+    span
+}
+
+fn transport_name(transport: AppServerTransport) -> &'static str {
+    match transport {
+        AppServerTransport::Stdio => "stdio",
+        AppServerTransport::WebSocket { .. } => "websocket",
+    }
+}
+
+fn client_name<'a>(
+    initialize_client_info: Option<&'a InitializeParams>,
+    session: &'a ConnectionSessionState,
+) -> Option<&'a str> {
+    if let Some(params) = initialize_client_info {
+        return Some(params.client_info.name.as_str());
+    }
+    session.app_server_client_name.as_deref()
+}
+
+fn client_version<'a>(
+    initialize_client_info: Option<&'a InitializeParams>,
+    session: &'a ConnectionSessionState,
+) -> Option<&'a str> {
+    if let Some(params) = initialize_client_info {
+        return Some(params.client_info.version.as_str());
+    }
+    session.client_version.as_deref()
+}
+
+fn initialize_client_info(request: &JSONRPCRequest) -> Option<InitializeParams> {
+    if request.method != "initialize" {
+        return None;
+    }
+    let params = request.params.clone()?;
+    serde_json::from_value(params).ok()
+}

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

@@ -6,6 +6,8 @@ use crate::error_code::INTERNAL_ERROR_CODE;
 use crate::error_code::INVALID_REQUEST_ERROR_CODE;
 use crate::outgoing_message::ClientRequestResult;
 use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
+use crate::server_request_error::is_turn_transition_server_request_error;
+use crate::thread_state::ThreadListenerCommand;
 use crate::thread_state::ThreadState;
 use crate::thread_state::TurnSummary;
 use crate::thread_status::ThreadWatchActiveGuard;
@@ -56,6 +58,7 @@ use codex_app_server_protocol::RawResponseItemCompletedNotification;
 use codex_app_server_protocol::ReasoningSummaryPartAddedNotification;
 use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification;
 use codex_app_server_protocol::ReasoningTextDeltaNotification;
+use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ServerRequestPayload;
 use codex_app_server_protocol::TerminalInteractionNotification;
@@ -80,6 +83,7 @@ use codex_app_server_protocol::TurnError;
 use codex_app_server_protocol::TurnInterruptResponse;
 use codex_app_server_protocol::TurnPlanStep;
 use codex_app_server_protocol::TurnPlanUpdatedNotification;
+use codex_app_server_protocol::TurnStartedNotification;
 use codex_app_server_protocol::TurnStatus;
 use codex_app_server_protocol::build_turns_from_rollout_items;
 use codex_app_server_protocol::convert_patch_changes;
@@ -132,6 +136,38 @@ struct CommandExecutionCompletionItem {
     command_actions: Vec<V2ParsedCommand>,
 }
 
+async fn resolve_server_request_on_thread_listener(
+    thread_state: &Arc<Mutex<ThreadState>>,
+    request_id: RequestId,
+) {
+    let (completion_tx, completion_rx) = oneshot::channel();
+    let listener_command_tx = {
+        let state = thread_state.lock().await;
+        state.listener_command_tx()
+    };
+    let Some(listener_command_tx) = listener_command_tx else {
+        error!("failed to remove pending client request: thread listener is not running");
+        return;
+    };
+
+    if listener_command_tx
+        .send(ThreadListenerCommand::ResolveServerRequest {
+            request_id,
+            completion_tx,
+        })
+        .is_err()
+    {
+        error!(
+            "failed to remove pending client request: thread listener command channel is closed"
+        );
+        return;
+    }
+
+    if let Err(err) = completion_rx.await {
+        error!("failed to remove pending client request: {err}");
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn apply_bespoke_event_handling(
     event: Event,
@@ -150,12 +186,34 @@ pub(crate) async fn apply_bespoke_event_handling(
         msg,
     } = event;
     match msg {
-        EventMsg::TurnStarted(_) => {
+        EventMsg::TurnStarted(payload) => {
+            // While not technically necessary as it was already done on TurnComplete, be extra cautios and abort any pending server requests.
+            outgoing.abort_pending_server_requests().await;
             thread_watch_manager
                 .note_turn_started(&conversation_id.to_string())
                 .await;
+            if let ApiVersion::V2 = api_version {
+                let turn = {
+                    let state = thread_state.lock().await;
+                    state.active_turn_snapshot().unwrap_or_else(|| Turn {
+                        id: payload.turn_id.clone(),
+                        items: Vec::new(),
+                        error: None,
+                        status: TurnStatus::InProgress,
+                    })
+                };
+                let notification = TurnStartedNotification {
+                    thread_id: conversation_id.to_string(),
+                    turn,
+                };
+                outgoing
+                    .send_server_notification(ServerNotification::TurnStarted(notification))
+                    .await;
+            }
         }
         EventMsg::TurnComplete(_ev) => {
+            // All per-thread requests are bound to a turn, so abort them.
+            outgoing.abort_pending_server_requests().await;
             let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some();
             thread_watch_manager
                 .note_turn_completed(&conversation_id.to_string(), turn_failed)
@@ -193,7 +251,6 @@ pub(crate) async fn apply_bespoke_event_handling(
         EventMsg::RealtimeConversationRealtime(event) => {
             if let ApiVersion::V2 = api_version {
                 match event.payload {
-                    RealtimeEvent::SessionCreated { .. } => {}
                     RealtimeEvent::SessionUpdated { .. } => {}
                     RealtimeEvent::AudioOut(audio) => {
                         let notification = ThreadRealtimeOutputAudioDeltaNotification {
@@ -217,6 +274,24 @@ pub(crate) async fn apply_bespoke_event_handling(
                             ))
                             .await;
                     }
+                    RealtimeEvent::ConversationItemDone { .. } => {}
+                    RealtimeEvent::HandoffRequested(handoff) => {
+                        let notification = ThreadRealtimeItemAddedNotification {
+                            thread_id: conversation_id.to_string(),
+                            item: serde_json::json!({
+                                "type": "handoff_request",
+                                "handoff_id": handoff.handoff_id,
+                                "item_id": handoff.item_id,
+                                "input_transcript": handoff.input_transcript,
+                                "messages": handoff.messages,
+                            }),
+                        };
+                        outgoing
+                            .send_server_notification(ServerNotification::ThreadRealtimeItemAdded(
+                                notification,
+                            ))
+                            .await;
+                    }
                     RealtimeEvent::Error(message) => {
                         let notification = ThreadRealtimeErrorNotification {
                             thread_id: conversation_id.to_string(),
@@ -263,7 +338,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                         reason,
                         grant_root,
                     };
-                    let rx = outgoing
+                    let (_pending_request_id, rx) = outgoing
                         .send_request(ServerRequestPayload::ApplyPatchApproval(params))
                         .await;
                     tokio::spawn(async move {
@@ -307,7 +382,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                         reason,
                         grant_root,
                     };
-                    let rx = outgoing
+                    let (pending_request_id, rx) = outgoing
                         .send_request(ServerRequestPayload::FileChangeRequestApproval(params))
                         .await;
                     tokio::spawn(async move {
@@ -316,6 +391,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                             conversation_id,
                             item_id,
                             patch_changes,
+                            pending_request_id,
                             rx,
                             conversation,
                             outgoing,
@@ -362,7 +438,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                         reason,
                         parsed_cmd,
                     };
-                    let rx = outgoing
+                    let (_pending_request_id, rx) = outgoing
                         .send_request(ServerRequestPayload::ExecCommandApproval(params))
                         .await;
                     tokio::spawn(async move {
@@ -435,7 +511,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                         proposed_network_policy_amendments: proposed_network_policy_amendments_v2,
                         available_decisions: Some(available_decisions),
                     };
-                    let rx = outgoing
+                    let (pending_request_id, rx) = outgoing
                         .send_request(ServerRequestPayload::CommandExecutionRequestApproval(
                             params,
                         ))
@@ -447,6 +523,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                             approval_id,
                             call_id,
                             completion_item,
+                            pending_request_id,
                             rx,
                             conversation,
                             outgoing,
@@ -489,14 +566,16 @@ pub(crate) async fn apply_bespoke_event_handling(
                     item_id: request.call_id,
                     questions,
                 };
-                let rx = outgoing
+                let (pending_request_id, rx) = outgoing
                     .send_request(ServerRequestPayload::ToolRequestUserInput(params))
                     .await;
                 tokio::spawn(async move {
                     on_request_user_input_response(
                         event_turn_id,
+                        pending_request_id,
                         rx,
                         conversation,
+                        thread_state,
                         user_input_guard,
                     )
                     .await;
@@ -550,7 +629,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                     tool: tool.clone(),
                     arguments: arguments.clone(),
                 };
-                let rx = outgoing
+                let (_pending_request_id, rx) = outgoing
                     .send_request(ServerRequestPayload::DynamicToolCall(params))
                     .await;
                 tokio::spawn(async move {
@@ -1136,6 +1215,7 @@ pub(crate) async fn apply_bespoke_event_handling(
             // Until we migrate the core to be aware of a first class FileChangeItem
             // and emit the corresponding EventMsg, we repurpose the call_id as the item_id.
             let item_id = patch_begin_event.call_id.clone();
+            let changes = convert_patch_changes(&patch_begin_event.changes);
 
             let first_start = {
                 let mut state = thread_state.lock().await;
@@ -1147,7 +1227,7 @@ pub(crate) async fn apply_bespoke_event_handling(
             if first_start {
                 let item = ThreadItem::FileChange {
                     id: item_id.clone(),
-                    changes: convert_patch_changes(&patch_begin_event.changes),
+                    changes,
                     status: PatchApplyStatus::InProgress,
                 };
                 let notification = ItemStartedNotification {
@@ -1329,6 +1409,8 @@ pub(crate) async fn apply_bespoke_event_handling(
         }
         // If this is a TurnAborted, reply to any pending interrupt requests.
         EventMsg::TurnAborted(turn_aborted_event) => {
+            // All per-thread requests are bound to a turn, so abort them.
+            outgoing.abort_pending_server_requests().await;
             let pending = {
                 let mut state = thread_state.lock().await;
                 std::mem::take(&mut state.pending_interrupts)
@@ -1725,6 +1807,7 @@ async fn on_patch_approval_response(
     let response = receiver.await;
     let value = match response {
         Ok(Ok(value)) => value,
+        Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
         Ok(Err(err)) => {
             error!("request failed with client error: {err:?}");
             if let Err(submit_err) = codex
@@ -1781,6 +1864,7 @@ async fn on_exec_approval_response(
     let response = receiver.await;
     let value = match response {
         Ok(Ok(value)) => value,
+        Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
         Ok(Err(err)) => {
             error!("request failed with client error: {err:?}");
             return;
@@ -1816,14 +1900,18 @@ async fn on_exec_approval_response(
 
 async fn on_request_user_input_response(
     event_turn_id: String,
+    pending_request_id: RequestId,
     receiver: oneshot::Receiver<ClientRequestResult>,
     conversation: Arc<CodexThread>,
+    thread_state: Arc<Mutex<ThreadState>>,
     user_input_guard: ThreadWatchActiveGuard,
 ) {
     let response = receiver.await;
+    resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
     drop(user_input_guard);
     let value = match response {
         Ok(Ok(value)) => value,
+        Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
         Ok(Err(err)) => {
             error!("request failed with client error: {err:?}");
             let empty = CoreRequestUserInputResponse {
@@ -1934,6 +2022,7 @@ async fn on_file_change_request_approval_response(
     conversation_id: ThreadId,
     item_id: String,
     changes: Vec<FileUpdateChange>,
+    pending_request_id: RequestId,
     receiver: oneshot::Receiver<ClientRequestResult>,
     codex: Arc<CodexThread>,
     outgoing: ThreadScopedOutgoingMessageSender,
@@ -1941,6 +2030,7 @@ async fn on_file_change_request_approval_response(
     permission_guard: ThreadWatchActiveGuard,
 ) {
     let response = receiver.await;
+    resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
     drop(permission_guard);
     let (decision, completion_status) = match response {
         Ok(Ok(value)) => {
@@ -1958,6 +2048,7 @@ async fn on_file_change_request_approval_response(
             // Only short-circuit on declines/cancels/failures.
             (decision, completion_status)
         }
+        Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
         Ok(Err(err)) => {
             error!("request failed with client error: {err:?}");
             (ReviewDecision::Denied, Some(PatchApplyStatus::Failed))
@@ -1999,6 +2090,7 @@ async fn on_command_execution_request_approval_response(
     approval_id: Option<String>,
     item_id: String,
     completion_item: Option<CommandExecutionCompletionItem>,
+    pending_request_id: RequestId,
     receiver: oneshot::Receiver<ClientRequestResult>,
     conversation: Arc<CodexThread>,
     outgoing: ThreadScopedOutgoingMessageSender,
@@ -2006,6 +2098,7 @@ async fn on_command_execution_request_approval_response(
     permission_guard: ThreadWatchActiveGuard,
 ) {
     let response = receiver.await;
+    resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
     drop(permission_guard);
     let (decision, completion_status) = match response {
         Ok(Ok(value)) => {
@@ -2057,6 +2150,7 @@ async fn on_command_execution_request_approval_response(
             };
             (decision, completion_status)
         }
+        Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
         Ok(Err(err)) => {
             error!("request failed with client error: {err:?}");
             (ReviewDecision::Denied, Some(CommandExecutionStatus::Failed))

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

@@ -9,6 +9,7 @@ use tokio::sync::oneshot;
 use tracing::error;
 
 use crate::outgoing_message::ClientRequestResult;
+use crate::server_request_error::is_turn_transition_server_request_error;
 
 pub(crate) async fn on_call_response(
     call_id: String,
@@ -18,6 +19,7 @@ pub(crate) async fn on_call_response(
     let response = receiver.await;
     let (response, _error) = match response {
         Ok(Ok(value)) => decode_response(value),
+        Ok(Err(err)) if is_turn_transition_server_request_error(&err) => return,
         Ok(Err(err)) => {
             error!("request failed with client error: {err:?}");
             fallback_response("dynamic tool request failed")

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

@@ -52,6 +52,7 @@ use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::registry::Registry;
 use tracing_subscriber::util::SubscriberInitExt;
 
+mod app_server_tracing;
 mod bespoke_event_handling;
 mod codex_message_processor;
 mod config_api;
@@ -63,6 +64,7 @@ mod fuzzy_file_search;
 mod message_processor;
 mod models;
 mod outgoing_message;
+mod server_request_error;
 mod thread_state;
 mod thread_status;
 mod transport;
@@ -446,7 +448,7 @@ pub async fn run_main_with_transport(
     let otel = codex_core::otel_init::build_provider(
         &config,
         env!("CARGO_PKG_VERSION"),
-        Some("codex_app_server"),
+        Some("codex-app-server"),
         default_analytics_enabled,
     )
     .map_err(|e| {
@@ -556,7 +558,6 @@ pub async fn run_main_with_transport(
             outgoing: outgoing_message_sender,
             arg0_paths,
             config: Arc::new(config),
-            single_client_mode,
             cli_overrides,
             loader_overrides,
             cloud_requirements: cloud_requirements.clone(),
@@ -674,6 +675,7 @@ pub async fn run_main_with_transport(
                                             .process_request(
                                                 connection_id,
                                                 request,
+                                                transport,
                                                 &mut connection_state.session,
                                                 &connection_state.outbound_initialized,
                                             )

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

@@ -12,6 +12,7 @@ use crate::external_agent_config_api::ExternalAgentConfigApi;
 use crate::outgoing_message::ConnectionId;
 use crate::outgoing_message::ConnectionRequestId;
 use crate::outgoing_message::OutgoingMessageSender;
+use crate::transport::AppServerTransport;
 use async_trait::async_trait;
 use codex_app_server_protocol::ChatgptAuthTokensRefreshParams;
 use codex_app_server_protocol::ChatgptAuthTokensRefreshReason;
@@ -59,6 +60,7 @@ use tokio::sync::watch;
 use tokio::time::Duration;
 use tokio::time::timeout;
 use toml::Value as TomlValue;
+use tracing::Instrument;
 
 const EXTERNAL_AUTH_REFRESH_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -140,13 +142,14 @@ pub(crate) struct ConnectionSessionState {
     pub(crate) initialized: bool,
     pub(crate) experimental_api_enabled: bool,
     pub(crate) opted_out_notification_methods: HashSet<String>,
+    pub(crate) app_server_client_name: Option<String>,
+    pub(crate) client_version: Option<String>,
 }
 
 pub(crate) struct MessageProcessorArgs {
     pub(crate) outgoing: Arc<OutgoingMessageSender>,
     pub(crate) arg0_paths: Arg0DispatchPaths,
     pub(crate) config: Arc<Config>,
-    pub(crate) single_client_mode: bool,
     pub(crate) cli_overrides: Vec<(String, TomlValue)>,
     pub(crate) loader_overrides: LoaderOverrides,
     pub(crate) cloud_requirements: CloudRequirementsLoader,
@@ -162,7 +165,6 @@ impl MessageProcessor {
             outgoing,
             arg0_paths,
             config,
-            single_client_mode,
             cli_overrides,
             loader_overrides,
             cloud_requirements,
@@ -198,7 +200,6 @@ impl MessageProcessor {
             config: Arc::clone(&config),
             cli_overrides: cli_overrides.clone(),
             cloud_requirements: cloud_requirements.clone(),
-            single_client_mode,
             feedback,
         });
         let config_api = ConfigApi::new(
@@ -223,46 +224,50 @@ impl MessageProcessor {
         &mut self,
         connection_id: ConnectionId,
         request: JSONRPCRequest,
+        transport: AppServerTransport,
         session: &mut ConnectionSessionState,
         outbound_initialized: &AtomicBool,
     ) {
-        let request_method = request.method.as_str();
-        tracing::trace!(
-            ?connection_id,
-            request_id = ?request.id,
-            "app-server request: {request_method}"
-        );
-        let request_id = ConnectionRequestId {
-            connection_id,
-            request_id: request.id.clone(),
-        };
-        let request_json = match serde_json::to_value(&request) {
-            Ok(request_json) => request_json,
-            Err(err) => {
-                let error = JSONRPCErrorError {
-                    code: INVALID_REQUEST_ERROR_CODE,
-                    message: format!("Invalid request: {err}"),
-                    data: None,
-                };
-                self.outgoing.send_error(request_id, error).await;
-                return;
-            }
-        };
+        let request_span =
+            crate::app_server_tracing::request_span(&request, transport, connection_id, session);
+        async {
+            let request_method = request.method.as_str();
+            tracing::trace!(
+                ?connection_id,
+                request_id = ?request.id,
+                "app-server request: {request_method}"
+            );
+            let request_id = ConnectionRequestId {
+                connection_id,
+                request_id: request.id.clone(),
+            };
+            let request_json = match serde_json::to_value(&request) {
+                Ok(request_json) => request_json,
+                Err(err) => {
+                    let error = JSONRPCErrorError {
+                        code: INVALID_REQUEST_ERROR_CODE,
+                        message: format!("Invalid request: {err}"),
+                        data: None,
+                    };
+                    self.outgoing.send_error(request_id, error).await;
+                    return;
+                }
+            };
 
-        let codex_request = match serde_json::from_value::<ClientRequest>(request_json) {
-            Ok(codex_request) => codex_request,
-            Err(err) => {
-                let error = JSONRPCErrorError {
-                    code: INVALID_REQUEST_ERROR_CODE,
-                    message: format!("Invalid request: {err}"),
-                    data: None,
-                };
-                self.outgoing.send_error(request_id, error).await;
-                return;
-            }
-        };
+            let codex_request = match serde_json::from_value::<ClientRequest>(request_json) {
+                Ok(codex_request) => codex_request,
+                Err(err) => {
+                    let error = JSONRPCErrorError {
+                        code: INVALID_REQUEST_ERROR_CODE,
+                        message: format!("Invalid request: {err}"),
+                        data: None,
+                    };
+                    self.outgoing.send_error(request_id, error).await;
+                    return;
+                }
+            };
 
-        match codex_request {
+            match codex_request {
             // Handle Initialize internally so CodexMessageProcessor does not have to concern
             // itself with the `initialized` bool.
             ClientRequest::Initialize { request_id, params } => {
@@ -303,6 +308,8 @@ impl MessageProcessor {
                         title: _title,
                         version,
                     } = params.client_info;
+                    session.app_server_client_name = Some(name.clone());
+                    session.client_version = Some(version.clone());
                     if let Err(error) = set_default_originator(name.clone()) {
                         match error {
                             SetOriginatorError::InvalidHeaderValue => {
@@ -336,6 +343,9 @@ impl MessageProcessor {
 
                     session.initialized = true;
                     outbound_initialized.store(true, Ordering::Release);
+                    self.codex_message_processor
+                        .connection_initialized(connection_id)
+                        .await;
                     return;
                 }
             }
@@ -350,91 +360,97 @@ impl MessageProcessor {
                     return;
                 }
             }
-        }
+            }
+            if let Some(reason) = codex_request.experimental_reason()
+                && !session.experimental_api_enabled
+            {
+                let error = JSONRPCErrorError {
+                    code: INVALID_REQUEST_ERROR_CODE,
+                    message: experimental_required_message(reason),
+                    data: None,
+                };
+                self.outgoing.send_error(request_id, error).await;
+                return;
+            }
 
-        if let Some(reason) = codex_request.experimental_reason()
-            && !session.experimental_api_enabled
-        {
-            let error = JSONRPCErrorError {
-                code: INVALID_REQUEST_ERROR_CODE,
-                message: experimental_required_message(reason),
-                data: None,
-            };
-            self.outgoing.send_error(request_id, error).await;
-            return;
-        }
-
-        match codex_request {
-            ClientRequest::ConfigRead { request_id, params } => {
-                self.handle_config_read(
-                    ConnectionRequestId {
-                        connection_id,
-                        request_id,
-                    },
-                    params,
-                )
-                .await;
-            }
-            ClientRequest::ExternalAgentConfigDetect { request_id, params } => {
-                self.handle_external_agent_config_detect(
-                    ConnectionRequestId {
-                        connection_id,
-                        request_id,
-                    },
-                    params,
-                )
-                .await;
-            }
-            ClientRequest::ExternalAgentConfigImport { request_id, params } => {
-                self.handle_external_agent_config_import(
-                    ConnectionRequestId {
-                        connection_id,
-                        request_id,
-                    },
-                    params,
-                )
-                .await;
-            }
-            ClientRequest::ConfigValueWrite { request_id, params } => {
-                self.handle_config_value_write(
-                    ConnectionRequestId {
-                        connection_id,
-                        request_id,
-                    },
-                    params,
-                )
-                .await;
-            }
-            ClientRequest::ConfigBatchWrite { request_id, params } => {
-                self.handle_config_batch_write(
-                    ConnectionRequestId {
-                        connection_id,
-                        request_id,
-                    },
-                    params,
-                )
-                .await;
-            }
-            ClientRequest::ConfigRequirementsRead {
-                request_id,
-                params: _,
-            } => {
-                self.handle_config_requirements_read(ConnectionRequestId {
-                    connection_id,
-                    request_id,
-                })
-                .await;
-            }
-            other => {
-                // Box the delegated future so this wrapper's async state machine does not
-                // inline the full `CodexMessageProcessor::process_request` future, which
-                // can otherwise push worker-thread stack usage over the edge.
-                self.codex_message_processor
-                    .process_request(connection_id, other)
-                    .boxed()
+            match codex_request {
+                ClientRequest::ConfigRead { request_id, params } => {
+                    self.handle_config_read(
+                        ConnectionRequestId {
+                            connection_id,
+                            request_id,
+                        },
+                        params,
+                    )
                     .await;
+                }
+                ClientRequest::ExternalAgentConfigDetect { request_id, params } => {
+                    self.handle_external_agent_config_detect(
+                        ConnectionRequestId {
+                            connection_id,
+                            request_id,
+                        },
+                        params,
+                    )
+                    .await;
+                }
+                ClientRequest::ExternalAgentConfigImport { request_id, params } => {
+                    self.handle_external_agent_config_import(
+                        ConnectionRequestId {
+                            connection_id,
+                            request_id,
+                        },
+                        params,
+                    )
+                    .await;
+                }
+                ClientRequest::ConfigValueWrite { request_id, params } => {
+                    self.handle_config_value_write(
+                        ConnectionRequestId {
+                            connection_id,
+                            request_id,
+                        },
+                        params,
+                    )
+                    .await;
+                }
+                ClientRequest::ConfigBatchWrite { request_id, params } => {
+                    self.handle_config_batch_write(
+                        ConnectionRequestId {
+                            connection_id,
+                            request_id,
+                        },
+                        params,
+                    )
+                    .await;
+                }
+                ClientRequest::ConfigRequirementsRead {
+                    request_id,
+                    params: _,
+                } => {
+                    self.handle_config_requirements_read(ConnectionRequestId {
+                        connection_id,
+                        request_id,
+                    })
+                    .await;
+                }
+                other => {
+                    // Box the delegated future so this wrapper's async state machine does not
+                    // inline the full `CodexMessageProcessor::process_request` future, which
+                    // can otherwise push worker-thread stack usage over the edge.
+                    self.codex_message_processor
+                        .process_request(
+                            connection_id,
+                            other,
+                            session.app_server_client_name.clone(),
+                        )
+                        .boxed()
+                        .await;
+                }
             }
         }
+        .instrument(request_span)
+        .await;
     }
 
     pub(crate) async fn process_notification(&self, notification: JSONRPCNotification) {

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

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use codex_app_server_protocol::Model;
+use codex_app_server_protocol::ModelUpgradeInfo;
 use codex_app_server_protocol::ReasoningEffortOption;
 use codex_core::ThreadManager;
 use codex_core::models_manager::manager::RefreshStrategy;
@@ -24,7 +25,14 @@ fn model_from_preset(preset: ModelPreset) -> Model {
     Model {
         id: preset.id.to_string(),
         model: preset.model.to_string(),
-        upgrade: preset.upgrade.map(|upgrade| upgrade.id),
+        upgrade: preset.upgrade.as_ref().map(|upgrade| upgrade.id.clone()),
+        upgrade_info: preset.upgrade.as_ref().map(|upgrade| ModelUpgradeInfo {
+            model: upgrade.id.clone(),
+            upgrade_copy: upgrade.upgrade_copy.clone(),
+            model_link: upgrade.model_link.clone(),
+            migration_markdown: upgrade.migration_markdown.clone(),
+        }),
+        availability_nux: preset.availability_nux.map(Into::into),
         display_name: preset.display_name.to_string(),
         description: preset.description.to_string(),
         hidden: !preset.show_in_picker,

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

@@ -17,6 +17,7 @@ use tokio::sync::oneshot;
 use tracing::warn;
 
 use crate::error_code::INTERNAL_ERROR_CODE;
+use crate::server_request_error::TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON;
 
 #[cfg(test)]
 use codex_protocol::account::PlanType;
@@ -62,6 +63,7 @@ pub(crate) struct ThreadScopedOutgoingMessageSender {
 struct PendingCallbackEntry {
     callback: oneshot::Sender<ClientRequestResult>,
     thread_id: Option<ThreadId>,
+    request: ServerRequest,
 }
 
 impl ThreadScopedOutgoingMessageSender {
@@ -80,12 +82,12 @@ impl ThreadScopedOutgoingMessageSender {
     pub(crate) async fn send_request(
         &self,
         payload: ServerRequestPayload,
-    ) -> oneshot::Receiver<ClientRequestResult> {
+    ) -> (RequestId, oneshot::Receiver<ClientRequestResult>) {
         self.outgoing
-            .send_request_to_thread_connections(
-                self.thread_id,
-                self.connection_ids.as_slice(),
+            .send_request_to_connections(
+                Some(self.connection_ids.as_slice()),
                 payload,
+                Some(self.thread_id),
             )
             .await
     }
@@ -99,6 +101,20 @@ impl ThreadScopedOutgoingMessageSender {
             .await;
     }
 
+    pub(crate) async fn abort_pending_server_requests(&self) {
+        self.outgoing
+            .cancel_requests_for_thread(
+                self.thread_id,
+                Some(JSONRPCErrorError {
+                    code: INTERNAL_ERROR_CODE,
+                    message: "client request resolved because the turn state was changed"
+                        .to_string(),
+                    data: Some(serde_json::json!({ "reason": TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON })),
+                }),
+            )
+            .await
+    }
+
     pub(crate) async fn send_response<T: Serialize>(
         &self,
         request_id: ConnectionRequestId,
@@ -129,38 +145,23 @@ impl OutgoingMessageSender {
         &self,
         request: ServerRequestPayload,
     ) -> (RequestId, oneshot::Receiver<ClientRequestResult>) {
-        self.send_request_with_id_to_connections(&[], request, None)
-            .await
-    }
-
-    async fn send_request_to_thread_connections(
-        &self,
-        thread_id: ThreadId,
-        connection_ids: &[ConnectionId],
-        request: ServerRequestPayload,
-    ) -> oneshot::Receiver<ClientRequestResult> {
-        if connection_ids.is_empty() {
-            let (_tx, rx) = oneshot::channel();
-            return rx;
-        }
-        let (_request_id, receiver) = self
-            .send_request_with_id_to_connections(connection_ids, request, Some(thread_id))
-            .await;
-        receiver
+        self.send_request_to_connections(None, request, None).await
     }
 
     fn next_request_id(&self) -> RequestId {
         RequestId::Integer(self.next_server_request_id.fetch_add(1, Ordering::Relaxed))
     }
 
-    async fn send_request_with_id_to_connections(
+    async fn send_request_to_connections(
         &self,
-        connection_ids: &[ConnectionId],
+        connection_ids: Option<&[ConnectionId]>,
         request: ServerRequestPayload,
         thread_id: Option<ThreadId>,
     ) -> (RequestId, oneshot::Receiver<ClientRequestResult>) {
         let id = self.next_request_id();
         let outgoing_message_id = id.clone();
+        let request = request.request_with_id(outgoing_message_id.clone());
+
         let (tx_approve, rx_approve) = oneshot::channel();
         {
             let mut request_id_to_callback = self.request_id_to_callback.lock().await;
@@ -169,36 +170,39 @@ impl OutgoingMessageSender {
                 PendingCallbackEntry {
                     callback: tx_approve,
                     thread_id,
+                    request: request.clone(),
                 },
             );
         }
 
-        let outgoing_message =
-            OutgoingMessage::Request(request.request_with_id(outgoing_message_id.clone()));
-        let send_result = if connection_ids.is_empty() {
-            self.sender
-                .send(OutgoingEnvelope::Broadcast {
-                    message: outgoing_message,
-                })
-                .await
-        } else {
-            let mut send_error = None;
-            for connection_id in connection_ids {
-                if let Err(err) = self
-                    .sender
-                    .send(OutgoingEnvelope::ToConnection {
-                        connection_id: *connection_id,
-                        message: outgoing_message.clone(),
+        let outgoing_message = OutgoingMessage::Request(request);
+        let send_result = match connection_ids {
+            None => {
+                self.sender
+                    .send(OutgoingEnvelope::Broadcast {
+                        message: outgoing_message,
                     })
                     .await
-                {
-                    send_error = Some(err);
-                    break;
-                }
             }
-            match send_error {
-                Some(err) => Err(err),
-                None => Ok(()),
+            Some(connection_ids) => {
+                let mut send_error = None;
+                for connection_id in connection_ids {
+                    if let Err(err) = self
+                        .sender
+                        .send(OutgoingEnvelope::ToConnection {
+                            connection_id: *connection_id,
+                            message: outgoing_message.clone(),
+                        })
+                        .await
+                    {
+                        send_error = Some(err);
+                        break;
+                    }
+                }
+                match send_error {
+                    Some(err) => Err(err),
+                    None => Ok(()),
+                }
             }
         };
 
@@ -210,11 +214,28 @@ impl OutgoingMessageSender {
         (outgoing_message_id, rx_approve)
     }
 
+    pub(crate) async fn replay_requests_to_connection_for_thread(
+        &self,
+        connection_id: ConnectionId,
+        thread_id: ThreadId,
+    ) {
+        let requests = self.pending_requests_for_thread(thread_id).await;
+        for request in requests {
+            if let Err(err) = self
+                .sender
+                .send(OutgoingEnvelope::ToConnection {
+                    connection_id,
+                    message: OutgoingMessage::Request(request),
+                })
+                .await
+            {
+                warn!("failed to resend request to client: {err:?}");
+            }
+        }
+    }
+
     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)
-        };
+        let entry = self.take_request_callback(&id).await;
 
         match entry {
             Some((id, entry)) => {
@@ -229,10 +250,7 @@ impl OutgoingMessageSender {
     }
 
     pub(crate) async fn notify_client_error(&self, id: RequestId, error: JSONRPCErrorError) {
-        let entry = {
-            let mut request_id_to_callback = self.request_id_to_callback.lock().await;
-            request_id_to_callback.remove_entry(&id)
-        };
+        let entry = self.take_request_callback(&id).await;
 
         match entry {
             Some((id, entry)) => {
@@ -248,23 +266,62 @@ impl OutgoingMessageSender {
     }
 
     pub(crate) async fn cancel_request(&self, id: &RequestId) -> bool {
-        let entry = {
-            let mut request_id_to_callback = self.request_id_to_callback.lock().await;
-            request_id_to_callback.remove_entry(id)
-        };
-        entry.is_some()
+        self.take_request_callback(id).await.is_some()
     }
 
-    pub(crate) async fn cancel_requests_for_thread(&self, thread_id: ThreadId) {
+    async fn take_request_callback(
+        &self,
+        id: &RequestId,
+    ) -> Option<(RequestId, PendingCallbackEntry)> {
         let mut request_id_to_callback = self.request_id_to_callback.lock().await;
-        let request_ids = request_id_to_callback
+        request_id_to_callback.remove_entry(id)
+    }
+
+    pub(crate) async fn pending_requests_for_thread(
+        &self,
+        thread_id: ThreadId,
+    ) -> Vec<ServerRequest> {
+        let request_id_to_callback = self.request_id_to_callback.lock().await;
+        let mut requests = request_id_to_callback
             .iter()
-            .filter_map(|(request_id, entry)| {
-                (entry.thread_id == Some(thread_id)).then_some(request_id.clone())
+            .filter_map(|(_, entry)| {
+                (entry.thread_id == Some(thread_id)).then_some(entry.request.clone())
             })
             .collect::<Vec<_>>();
-        for request_id in request_ids {
-            request_id_to_callback.remove(&request_id);
+        requests.sort_by(|left, right| left.id().cmp(right.id()));
+        requests
+    }
+
+    pub(crate) async fn cancel_requests_for_thread(
+        &self,
+        thread_id: ThreadId,
+        error: Option<JSONRPCErrorError>,
+    ) {
+        let entries = {
+            let mut request_id_to_callback = self.request_id_to_callback.lock().await;
+            let request_ids = request_id_to_callback
+                .iter()
+                .filter_map(|(request_id, entry)| {
+                    (entry.thread_id == Some(thread_id)).then_some(request_id.clone())
+                })
+                .collect::<Vec<_>>();
+
+            let mut entries = Vec::with_capacity(request_ids.len());
+            for request_id in request_ids {
+                if let Some(entry) = request_id_to_callback.remove(&request_id) {
+                    entries.push(entry);
+                }
+            }
+            entries
+        };
+
+        if let Some(error) = error {
+            for entry in entries {
+                if let Err(err) = entry.callback.send(Err(error.clone())) {
+                    let request_id = entry.request.id();
+                    warn!("could not notify callback for {request_id:?} due to: {err:?}",);
+                }
+            }
         }
     }
 
@@ -441,14 +498,18 @@ mod tests {
     use codex_app_server_protocol::ApplyPatchApprovalParams;
     use codex_app_server_protocol::AuthMode;
     use codex_app_server_protocol::ConfigWarningNotification;
+    use codex_app_server_protocol::DynamicToolCallParams;
+    use codex_app_server_protocol::FileChangeRequestApprovalParams;
     use codex_app_server_protocol::LoginChatGptCompleteNotification;
     use codex_app_server_protocol::ModelRerouteReason;
     use codex_app_server_protocol::ModelReroutedNotification;
     use codex_app_server_protocol::RateLimitSnapshot;
     use codex_app_server_protocol::RateLimitWindow;
+    use codex_app_server_protocol::ToolRequestUserInputParams;
     use codex_protocol::ThreadId;
     use pretty_assertions::assert_eq;
     use serde_json::json;
+    use std::sync::Arc;
     use tokio::time::timeout;
     use uuid::Uuid;
 
@@ -551,6 +612,7 @@ mod tests {
     fn verify_account_updated_notification_serialization() {
         let notification = ServerNotification::AccountUpdated(AccountUpdatedNotification {
             auth_mode: Some(AuthMode::ApiKey),
+            plan_type: None,
         });
 
         let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
@@ -558,7 +620,8 @@ mod tests {
             json!({
                 "method": "account/updated",
                 "params": {
-                    "authMode": "apikey"
+                    "authMode": "apikey",
+                    "planType": null
                 },
             }),
             serde_json::to_value(jsonrpc_notification)
@@ -723,4 +786,121 @@ mod tests {
             .expect("waiter should receive a callback");
         assert_eq!(result, Err(error));
     }
+
+    #[tokio::test]
+    async fn pending_requests_for_thread_returns_thread_requests_in_request_id_order() {
+        let (tx, _rx) = mpsc::channel::<OutgoingEnvelope>(8);
+        let outgoing = Arc::new(OutgoingMessageSender::new(tx));
+        let thread_id = ThreadId::new();
+        let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
+            outgoing.clone(),
+            vec![ConnectionId(1)],
+            thread_id,
+        );
+
+        let (dynamic_tool_request_id, _dynamic_tool_waiter) = thread_outgoing
+            .send_request(ServerRequestPayload::DynamicToolCall(
+                DynamicToolCallParams {
+                    thread_id: thread_id.to_string(),
+                    turn_id: "turn-1".to_string(),
+                    call_id: "call-0".to_string(),
+                    tool: "tool".to_string(),
+                    arguments: json!({}),
+                },
+            ))
+            .await;
+        let (first_request_id, _first_waiter) = thread_outgoing
+            .send_request(ServerRequestPayload::ToolRequestUserInput(
+                ToolRequestUserInputParams {
+                    thread_id: thread_id.to_string(),
+                    turn_id: "turn-1".to_string(),
+                    item_id: "call-1".to_string(),
+                    questions: vec![],
+                },
+            ))
+            .await;
+        let (second_request_id, _second_waiter) = thread_outgoing
+            .send_request(ServerRequestPayload::FileChangeRequestApproval(
+                FileChangeRequestApprovalParams {
+                    thread_id: thread_id.to_string(),
+                    turn_id: "turn-1".to_string(),
+                    item_id: "call-2".to_string(),
+                    reason: None,
+                    grant_root: None,
+                },
+            ))
+            .await;
+        let pending_requests = outgoing.pending_requests_for_thread(thread_id).await;
+        assert_eq!(
+            pending_requests
+                .iter()
+                .map(ServerRequest::id)
+                .collect::<Vec<_>>(),
+            vec![
+                &dynamic_tool_request_id,
+                &first_request_id,
+                &second_request_id
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn cancel_requests_for_thread_cancels_all_thread_requests() {
+        let (tx, _rx) = mpsc::channel::<OutgoingEnvelope>(8);
+        let outgoing = Arc::new(OutgoingMessageSender::new(tx));
+        let thread_id = ThreadId::new();
+        let thread_outgoing = ThreadScopedOutgoingMessageSender::new(
+            outgoing.clone(),
+            vec![ConnectionId(1)],
+            thread_id,
+        );
+
+        let (_dynamic_tool_request_id, dynamic_tool_waiter) = thread_outgoing
+            .send_request(ServerRequestPayload::DynamicToolCall(
+                DynamicToolCallParams {
+                    thread_id: thread_id.to_string(),
+                    turn_id: "turn-1".to_string(),
+                    call_id: "call-0".to_string(),
+                    tool: "tool".to_string(),
+                    arguments: json!({}),
+                },
+            ))
+            .await;
+        let (_request_id, user_input_waiter) = thread_outgoing
+            .send_request(ServerRequestPayload::ToolRequestUserInput(
+                ToolRequestUserInputParams {
+                    thread_id: thread_id.to_string(),
+                    turn_id: "turn-1".to_string(),
+                    item_id: "call-1".to_string(),
+                    questions: vec![],
+                },
+            ))
+            .await;
+        let error = JSONRPCErrorError {
+            code: INTERNAL_ERROR_CODE,
+            message: "tracked request cancelled".to_string(),
+            data: None,
+        };
+
+        outgoing
+            .cancel_requests_for_thread(thread_id, Some(error.clone()))
+            .await;
+
+        let dynamic_tool_result = timeout(Duration::from_secs(1), dynamic_tool_waiter)
+            .await
+            .expect("dynamic tool waiter should resolve")
+            .expect("dynamic tool waiter should receive a callback");
+        let user_input_result = timeout(Duration::from_secs(1), user_input_waiter)
+            .await
+            .expect("user input waiter should resolve")
+            .expect("user input waiter should receive a callback");
+        assert_eq!(dynamic_tool_result, Err(error.clone()));
+        assert_eq!(user_input_result, Err(error));
+        assert!(
+            outgoing
+                .pending_requests_for_thread(thread_id)
+                .await
+                .is_empty()
+        );
+    }
 }

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

@@ -0,0 +1,42 @@
+use codex_app_server_protocol::JSONRPCErrorError;
+
+pub(crate) const TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON: &str = "turnTransition";
+
+pub(crate) fn is_turn_transition_server_request_error(error: &JSONRPCErrorError) -> bool {
+    error
+        .data
+        .as_ref()
+        .and_then(|data| data.get("reason"))
+        .and_then(serde_json::Value::as_str)
+        == Some(TURN_TRANSITION_PENDING_REQUEST_ERROR_REASON)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_turn_transition_server_request_error;
+    use codex_app_server_protocol::JSONRPCErrorError;
+    use pretty_assertions::assert_eq;
+    use serde_json::json;
+
+    #[test]
+    fn turn_transition_error_is_detected() {
+        let error = JSONRPCErrorError {
+            code: -1,
+            message: "client request resolved because the turn state was changed".to_string(),
+            data: Some(json!({ "reason": "turnTransition" })),
+        };
+
+        assert_eq!(is_turn_transition_server_request_error(&error), true);
+    }
+
+    #[test]
+    fn unrelated_error_is_not_detected() {
+        let error = JSONRPCErrorError {
+            code: -1,
+            message: "boom".to_string(),
+            data: Some(json!({ "reason": "other" })),
+        };
+
+        assert_eq!(is_turn_transition_server_request_error(&error), false);
+    }
+}

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

@@ -1,5 +1,6 @@
 use crate::outgoing_message::ConnectionId;
 use crate::outgoing_message::ConnectionRequestId;
+use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadHistoryBuilder;
 use codex_app_server_protocol::Turn;
 use codex_app_server_protocol::TurnError;
@@ -28,8 +29,16 @@ pub(crate) struct PendingThreadResumeRequest {
     pub(crate) config_snapshot: ThreadConfigSnapshot,
 }
 
+// ThreadListenerCommand is used to perform operations in the context of the thread listener, for serialization purposes.
 pub(crate) enum ThreadListenerCommand {
-    SendThreadResumeResponse(PendingThreadResumeRequest),
+    // SendThreadResumeResponse is used to resume an already running thread by sending the thread's history to the client and atomically subscribing for new updates.
+    SendThreadResumeResponse(Box<PendingThreadResumeRequest>),
+    // ResolveServerRequest is used to notify the client that the request has been resolved.
+    // It is executed in the thread listener's context to ensure that the resolved notification is ordered with regard to the request itself.
+    ResolveServerRequest {
+        request_id: RequestId,
+        completion_tx: oneshot::Sender<()>,
+    },
 }
 
 /// Per-conversation accumulation of the latest states e.g. error message while a turn runs.
@@ -51,7 +60,6 @@ pub(crate) struct ThreadState {
     listener_command_tx: Option<mpsc::UnboundedSender<ThreadListenerCommand>>,
     current_turn_history: ThreadHistoryBuilder,
     listener_thread: Option<Weak<CodexThread>>,
-    subscribed_connections: HashSet<ConnectionId>,
 }
 
 impl ThreadState {
@@ -86,18 +94,6 @@ impl ThreadState {
         self.listener_thread = None;
     }
 
-    pub(crate) fn add_connection(&mut self, connection_id: ConnectionId) {
-        self.subscribed_connections.insert(connection_id);
-    }
-
-    pub(crate) fn remove_connection(&mut self, connection_id: ConnectionId) {
-        self.subscribed_connections.remove(&connection_id);
-    }
-
-    pub(crate) fn subscribed_connection_ids(&self) -> Vec<ConnectionId> {
-        self.subscribed_connections.iter().copied().collect()
-    }
-
     pub(crate) fn set_experimental_raw_events(&mut self, enabled: bool) {
         self.experimental_raw_events = enabled;
     }
@@ -126,55 +122,112 @@ struct SubscriptionState {
     connection_id: ConnectionId,
 }
 
+struct ThreadEntry {
+    state: Arc<Mutex<ThreadState>>,
+    connection_ids: HashSet<ConnectionId>,
+}
+
+impl Default for ThreadEntry {
+    fn default() -> Self {
+        Self {
+            state: Arc::new(Mutex::new(ThreadState::default())),
+            connection_ids: HashSet::new(),
+        }
+    }
+}
+
 #[derive(Default)]
-pub(crate) struct ThreadStateManager {
-    thread_states: HashMap<ThreadId, Arc<Mutex<ThreadState>>>,
+struct ThreadStateManagerInner {
+    live_connections: HashSet<ConnectionId>,
+    threads: HashMap<ThreadId, ThreadEntry>,
     subscription_state_by_id: HashMap<Uuid, SubscriptionState>,
     thread_ids_by_connection: HashMap<ConnectionId, HashSet<ThreadId>>,
 }
 
+#[derive(Clone, Default)]
+pub(crate) struct ThreadStateManager {
+    state: Arc<Mutex<ThreadStateManagerInner>>,
+}
+
 impl ThreadStateManager {
     pub(crate) fn new() -> Self {
         Self::default()
     }
 
-    pub(crate) fn thread_state(&mut self, thread_id: ThreadId) -> Arc<Mutex<ThreadState>> {
-        self.thread_states
-            .entry(thread_id)
-            .or_insert_with(|| Arc::new(Mutex::new(ThreadState::default())))
-            .clone()
+    pub(crate) async fn connection_initialized(&self, connection_id: ConnectionId) {
+        self.state
+            .lock()
+            .await
+            .live_connections
+            .insert(connection_id);
     }
 
-    pub(crate) async fn remove_listener(&mut self, subscription_id: Uuid) -> Option<ThreadId> {
-        let subscription_state = self.subscription_state_by_id.remove(&subscription_id)?;
+    pub(crate) async fn subscribed_connection_ids(&self, thread_id: ThreadId) -> Vec<ConnectionId> {
+        let state = self.state.lock().await;
+        state
+            .threads
+            .get(&thread_id)
+            .map(|thread_entry| thread_entry.connection_ids.iter().copied().collect())
+            .unwrap_or_default()
+    }
+
+    pub(crate) async fn thread_state(&self, thread_id: ThreadId) -> Arc<Mutex<ThreadState>> {
+        let mut state = self.state.lock().await;
+        state.threads.entry(thread_id).or_default().state.clone()
+    }
+
+    pub(crate) async fn remove_listener(&self, subscription_id: Uuid) -> Option<ThreadId> {
+        let (subscription_state, connection_still_subscribed_to_thread, thread_state) = {
+            let mut state = self.state.lock().await;
+            let subscription_state = state.subscription_state_by_id.remove(&subscription_id)?;
+            let thread_id = subscription_state.thread_id;
+
+            let connection_still_subscribed_to_thread = state
+                .subscription_state_by_id
+                .values()
+                .any(|subscription_state_entry| {
+                    subscription_state_entry.thread_id == thread_id
+                        && subscription_state_entry.connection_id
+                            == subscription_state.connection_id
+                });
+            if !connection_still_subscribed_to_thread {
+                let mut remove_connection_entry = false;
+                if let Some(thread_ids) = state
+                    .thread_ids_by_connection
+                    .get_mut(&subscription_state.connection_id)
+                {
+                    thread_ids.remove(&thread_id);
+                    remove_connection_entry = thread_ids.is_empty();
+                }
+                if remove_connection_entry {
+                    state
+                        .thread_ids_by_connection
+                        .remove(&subscription_state.connection_id);
+                }
+                if let Some(thread_entry) = state.threads.get_mut(&thread_id) {
+                    thread_entry
+                        .connection_ids
+                        .remove(&subscription_state.connection_id);
+                }
+            }
+
+            let thread_state = state.threads.get(&thread_id).map(|thread_entry| {
+                (
+                    thread_entry.connection_ids.is_empty(),
+                    thread_entry.state.clone(),
+                )
+            });
+            (
+                subscription_state,
+                connection_still_subscribed_to_thread,
+                thread_state,
+            )
+        };
         let thread_id = subscription_state.thread_id;
 
-        let connection_still_subscribed_to_thread =
-            self.subscription_state_by_id.values().any(|state| {
-                state.thread_id == thread_id
-                    && state.connection_id == subscription_state.connection_id
-            });
-        if !connection_still_subscribed_to_thread {
-            let mut remove_connection_entry = false;
-            if let Some(thread_ids) = self
-                .thread_ids_by_connection
-                .get_mut(&subscription_state.connection_id)
-            {
-                thread_ids.remove(&thread_id);
-                remove_connection_entry = thread_ids.is_empty();
-            }
-            if remove_connection_entry {
-                self.thread_ids_by_connection
-                    .remove(&subscription_state.connection_id);
-            }
-        }
-
-        if let Some(thread_state) = self.thread_states.get(&thread_id) {
-            let mut thread_state = thread_state.lock().await;
-            if !connection_still_subscribed_to_thread {
-                thread_state.remove_connection(subscription_state.connection_id);
-            }
-            if thread_state.subscribed_connection_ids().is_empty() {
+        if let Some((no_subscribers, thread_state)) = thread_state {
+            let thread_state = thread_state.lock().await;
+            if !connection_still_subscribed_to_thread && no_subscribers {
                 tracing::debug!(
                     thread_id = %thread_id,
                     subscription_id = %subscription_id,
@@ -187,8 +240,24 @@ impl ThreadStateManager {
         Some(thread_id)
     }
 
-    pub(crate) async fn remove_thread_state(&mut self, thread_id: ThreadId) {
-        if let Some(thread_state) = self.thread_states.remove(&thread_id) {
+    pub(crate) async fn remove_thread_state(&self, thread_id: ThreadId) {
+        let thread_state = {
+            let mut state = self.state.lock().await;
+            let thread_state = state
+                .threads
+                .remove(&thread_id)
+                .map(|thread_entry| thread_entry.state);
+            state
+                .subscription_state_by_id
+                .retain(|_, state| state.thread_id != thread_id);
+            state.thread_ids_by_connection.retain(|_, thread_ids| {
+                thread_ids.remove(&thread_id);
+                !thread_ids.is_empty()
+            });
+            thread_state
+        };
+
+        if let Some(thread_state) = thread_state {
             let mut thread_state = thread_state.lock().await;
             tracing::debug!(
                 thread_id = %thread_id,
@@ -199,142 +268,189 @@ impl ThreadStateManager {
             );
             thread_state.clear_listener();
         }
-        self.subscription_state_by_id
-            .retain(|_, state| state.thread_id != thread_id);
-        self.thread_ids_by_connection.retain(|_, thread_ids| {
-            thread_ids.remove(&thread_id);
-            !thread_ids.is_empty()
-        });
     }
 
     pub(crate) async fn unsubscribe_connection_from_thread(
-        &mut self,
+        &self,
         thread_id: ThreadId,
         connection_id: ConnectionId,
     ) -> bool {
-        let Some(thread_state) = self.thread_states.get(&thread_id) else {
-            return false;
+        {
+            let mut state = self.state.lock().await;
+            if !state.threads.contains_key(&thread_id) {
+                return false;
+            }
+
+            if !state
+                .thread_ids_by_connection
+                .get(&connection_id)
+                .is_some_and(|thread_ids| thread_ids.contains(&thread_id))
+            {
+                return false;
+            }
+
+            if let Some(thread_ids) = state.thread_ids_by_connection.get_mut(&connection_id) {
+                thread_ids.remove(&thread_id);
+                if thread_ids.is_empty() {
+                    state.thread_ids_by_connection.remove(&connection_id);
+                }
+            }
+            if let Some(thread_entry) = state.threads.get_mut(&thread_id) {
+                thread_entry.connection_ids.remove(&connection_id);
+            }
+
+            state
+                .subscription_state_by_id
+                .retain(|_, subscription_state| {
+                    !(subscription_state.thread_id == thread_id
+                        && subscription_state.connection_id == connection_id)
+                });
         };
 
-        if !self
-            .thread_ids_by_connection
-            .get(&connection_id)
-            .is_some_and(|thread_ids| thread_ids.contains(&thread_id))
-        {
-            return false;
-        }
-
-        if let Some(thread_ids) = self.thread_ids_by_connection.get_mut(&connection_id) {
-            thread_ids.remove(&thread_id);
-            if thread_ids.is_empty() {
-                self.thread_ids_by_connection.remove(&connection_id);
-            }
-        }
-
-        self.subscription_state_by_id.retain(|_, state| {
-            !(state.thread_id == thread_id && state.connection_id == connection_id)
-        });
-
-        let mut thread_state = thread_state.lock().await;
-        thread_state.remove_connection(connection_id);
         true
     }
 
     pub(crate) async fn has_subscribers(&self, thread_id: ThreadId) -> bool {
-        let Some(thread_state) = self.thread_states.get(&thread_id) else {
-            return false;
-        };
-        !thread_state
+        self.state
             .lock()
             .await
-            .subscribed_connection_ids()
-            .is_empty()
+            .threads
+            .get(&thread_id)
+            .is_some_and(|thread_entry| !thread_entry.connection_ids.is_empty())
     }
 
     pub(crate) async fn set_listener(
-        &mut self,
+        &self,
         subscription_id: Uuid,
         thread_id: ThreadId,
         connection_id: ConnectionId,
         experimental_raw_events: bool,
     ) -> Arc<Mutex<ThreadState>> {
-        self.subscription_state_by_id.insert(
-            subscription_id,
-            SubscriptionState {
-                thread_id,
-                connection_id,
-            },
-        );
-        self.thread_ids_by_connection
-            .entry(connection_id)
-            .or_default()
-            .insert(thread_id);
-        let thread_state = self.thread_state(thread_id);
+        let thread_state = {
+            let mut state = self.state.lock().await;
+            state.subscription_state_by_id.insert(
+                subscription_id,
+                SubscriptionState {
+                    thread_id,
+                    connection_id,
+                },
+            );
+            state
+                .thread_ids_by_connection
+                .entry(connection_id)
+                .or_default()
+                .insert(thread_id);
+            let thread_entry = state.threads.entry(thread_id).or_default();
+            thread_entry.connection_ids.insert(connection_id);
+            thread_entry.state.clone()
+        };
         {
             let mut thread_state_guard = thread_state.lock().await;
-            thread_state_guard.add_connection(connection_id);
             thread_state_guard.set_experimental_raw_events(experimental_raw_events);
         }
         thread_state
     }
 
-    pub(crate) async fn ensure_connection_subscribed(
-        &mut self,
+    pub(crate) async fn try_ensure_connection_subscribed(
+        &self,
         thread_id: ThreadId,
         connection_id: ConnectionId,
         experimental_raw_events: bool,
-    ) -> Arc<Mutex<ThreadState>> {
-        self.thread_ids_by_connection
-            .entry(connection_id)
-            .or_default()
-            .insert(thread_id);
-        let thread_state = self.thread_state(thread_id);
+    ) -> Option<Arc<Mutex<ThreadState>>> {
+        let thread_state = {
+            let mut state = self.state.lock().await;
+            if !state.live_connections.contains(&connection_id) {
+                return None;
+            }
+            state
+                .thread_ids_by_connection
+                .entry(connection_id)
+                .or_default()
+                .insert(thread_id);
+            let thread_entry = state.threads.entry(thread_id).or_default();
+            thread_entry.connection_ids.insert(connection_id);
+            thread_entry.state.clone()
+        };
         {
             let mut thread_state_guard = thread_state.lock().await;
-            thread_state_guard.add_connection(connection_id);
             if experimental_raw_events {
                 thread_state_guard.set_experimental_raw_events(true);
             }
         }
-        thread_state
+        Some(thread_state)
     }
 
-    pub(crate) async fn remove_connection(&mut self, connection_id: ConnectionId) {
-        let thread_ids = self
-            .thread_ids_by_connection
-            .remove(&connection_id)
-            .unwrap_or_default();
-        self.subscription_state_by_id
-            .retain(|_, state| state.connection_id != connection_id);
-
-        if thread_ids.is_empty() {
-            for thread_state in self.thread_states.values() {
-                let mut thread_state = thread_state.lock().await;
-                thread_state.remove_connection(connection_id);
-                if thread_state.subscribed_connection_ids().is_empty() {
-                    tracing::debug!(
-                        connection_id = ?connection_id,
-                        listener_generation = thread_state.listener_generation,
-                        "retaining thread listener after connection disconnect left zero subscribers"
-                    );
-                }
-            }
-            return;
+    pub(crate) async fn try_add_connection_to_thread(
+        &self,
+        thread_id: ThreadId,
+        connection_id: ConnectionId,
+    ) -> bool {
+        let mut state = self.state.lock().await;
+        if !state.live_connections.contains(&connection_id) {
+            return false;
         }
+        state
+            .thread_ids_by_connection
+            .entry(connection_id)
+            .or_default()
+            .insert(thread_id);
+        state
+            .threads
+            .entry(thread_id)
+            .or_default()
+            .connection_ids
+            .insert(connection_id);
+        true
+    }
 
-        for thread_id in thread_ids {
-            if let Some(thread_state) = self.thread_states.get(&thread_id) {
-                let mut thread_state = thread_state.lock().await;
-                thread_state.remove_connection(connection_id);
-                if thread_state.subscribed_connection_ids().is_empty() {
-                    tracing::debug!(
-                        thread_id = %thread_id,
-                        connection_id = ?connection_id,
-                        listener_generation = thread_state.listener_generation,
-                        "retaining thread listener after connection disconnect left zero subscribers"
-                    );
+    pub(crate) async fn remove_connection(&self, connection_id: ConnectionId) {
+        let thread_states = {
+            let mut state = self.state.lock().await;
+            state.live_connections.remove(&connection_id);
+            let thread_ids = state
+                .thread_ids_by_connection
+                .remove(&connection_id)
+                .unwrap_or_default();
+            state
+                .subscription_state_by_id
+                .retain(|_, state| state.connection_id != connection_id);
+            for thread_id in &thread_ids {
+                if let Some(thread_entry) = state.threads.get_mut(thread_id) {
+                    thread_entry.connection_ids.remove(&connection_id);
                 }
             }
+            thread_ids
+                .into_iter()
+                .map(|thread_id| {
+                    (
+                        thread_id,
+                        state
+                            .threads
+                            .get(&thread_id)
+                            .is_none_or(|thread_entry| thread_entry.connection_ids.is_empty()),
+                        state
+                            .threads
+                            .get(&thread_id)
+                            .map(|thread_entry| thread_entry.state.clone()),
+                    )
+                })
+                .collect::<Vec<_>>()
+        };
+
+        for (thread_id, no_subscribers, thread_state) in thread_states {
+            if !no_subscribers {
+                continue;
+            }
+            let Some(thread_state) = thread_state else {
+                continue;
+            };
+            let listener_generation = thread_state.lock().await.listener_generation;
+            tracing::debug!(
+                thread_id = %thread_id,
+                connection_id = ?connection_id,
+                listener_generation,
+                "retaining thread listener after connection disconnect left zero subscribers"
+            );
         }
     }
 }

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

@@ -91,7 +91,12 @@ impl ThreadWatchManager {
     }
 
     pub(crate) async fn upsert_thread(&self, thread: Thread) {
-        self.mutate_and_publish(move |state| state.upsert_thread(thread.id))
+        self.mutate_and_publish(move |state| state.upsert_thread(thread.id, true))
+            .await;
+    }
+
+    pub(crate) async fn upsert_thread_silently(&self, thread: Thread) {
+        self.mutate_and_publish(move |state| state.upsert_thread(thread.id, false))
             .await;
     }
 
@@ -289,14 +294,22 @@ struct ThreadWatchState {
 }
 
 impl ThreadWatchState {
-    fn upsert_thread(&mut self, thread_id: String) -> Option<ThreadStatusChangedNotification> {
+    fn upsert_thread(
+        &mut self,
+        thread_id: String,
+        emit_notification: bool,
+    ) -> Option<ThreadStatusChangedNotification> {
         let previous_status = self.status_for(&thread_id);
         let runtime = self
             .runtime_by_thread_id
             .entry(thread_id.clone())
             .or_default();
         runtime.is_loaded = true;
-        self.status_changed_notification(thread_id, previous_status)
+        if emit_notification {
+            self.status_changed_notification(thread_id, previous_status)
+        } else {
+            None
+        }
     }
 
     fn remove_thread(&mut self, thread_id: &str) -> Option<ThreadStatusChangedNotification> {
@@ -692,6 +705,45 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn silent_upsert_skips_initial_notification() {
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(8);
+        let manager = ThreadWatchManager::new_with_outgoing(Arc::new(OutgoingMessageSender::new(
+            outgoing_tx,
+        )));
+
+        manager
+            .upsert_thread_silently(test_thread(
+                INTERACTIVE_THREAD_ID,
+                codex_app_server_protocol::SessionSource::Cli,
+            ))
+            .await;
+
+        assert_eq!(
+            manager
+                .loaded_status_for_thread(INTERACTIVE_THREAD_ID)
+                .await,
+            ThreadStatus::Idle,
+        );
+        assert!(
+            timeout(Duration::from_millis(100), outgoing_rx.recv())
+                .await
+                .is_err(),
+            "silent upsert should not emit thread/status/changed"
+        );
+
+        manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
+        assert_eq!(
+            recv_status_changed_notification(&mut outgoing_rx).await,
+            ThreadStatusChangedNotification {
+                thread_id: INTERACTIVE_THREAD_ID.to_string(),
+                status: ThreadStatus::Active {
+                    active_flags: vec![],
+                },
+            },
+        );
+    }
+
     async fn wait_for_status(
         manager: &ThreadWatchManager,
         thread_id: &str,
@@ -733,6 +785,7 @@ mod tests {
         Thread {
             id: thread_id.to_string(),
             preview: String::new(),
+            ephemeral: false,
             model_provider: "mock-provider".to_string(),
             created_at: 0,
             updated_at: 0,

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

@@ -671,12 +671,17 @@ pub(crate) async fn route_outgoing_envelope(
 mod tests {
     use super::*;
     use crate::error_code::OVERLOADED_ERROR_CODE;
+    use codex_utils_absolute_path::AbsolutePathBuf;
     use pretty_assertions::assert_eq;
     use serde_json::json;
     use std::path::PathBuf;
     use tokio::time::Duration;
     use tokio::time::timeout;
 
+    fn absolute_path(path: &str) -> AbsolutePathBuf {
+        AbsolutePathBuf::from_absolute_path(path).expect("absolute path")
+    }
+
     #[test]
     fn app_server_transport_parses_stdio_listen_url() {
         let transport = AppServerTransport::from_listen_url(AppServerTransport::DEFAULT_LISTEN_URL)
@@ -739,6 +744,7 @@ mod tests {
             id: codex_app_server_protocol::RequestId::Integer(7),
             method: "config/read".to_string(),
             params: Some(json!({ "includeLayers": false })),
+            trace: None,
         });
         assert!(
             enqueue_incoming_message(&transport_event_tx, &writer_tx, connection_id, request).await
@@ -880,6 +886,7 @@ mod tests {
             id: codex_app_server_protocol::RequestId::Integer(7),
             method: "config/read".to_string(),
             params: Some(json!({ "includeLayers": false })),
+            trace: None,
         });
 
         let enqueue_result = tokio::time::timeout(
@@ -977,7 +984,7 @@ mod tests {
                                 network: None,
                                 file_system: Some(
                                     codex_app_server_protocol::AdditionalFileSystemPermissions {
-                                        read: Some(vec![PathBuf::from("/tmp/allowed")]),
+                                        read: Some(vec![absolute_path("/tmp/allowed")]),
                                         write: None,
                                     },
                                 ),
@@ -1039,7 +1046,7 @@ mod tests {
                                 network: None,
                                 file_system: Some(
                                     codex_app_server_protocol::AdditionalFileSystemPermissions {
-                                        read: Some(vec![PathBuf::from("/tmp/allowed")]),
+                                        read: Some(vec![absolute_path("/tmp/allowed")]),
                                         write: None,
                                     },
                                 ),
@@ -1060,12 +1067,13 @@ mod tests {
             .await
             .expect("request should be delivered to the connection");
         let json = serde_json::to_value(message).expect("request should serialize");
+        let allowed_path = absolute_path("/tmp/allowed").to_string_lossy().into_owned();
         assert_eq!(
             json["params"]["additionalPermissions"],
             json!({
                 "network": null,
                 "fileSystem": {
-                    "read": ["/tmp/allowed"],
+                    "read": [allowed_path],
                     "write": null,
                 },
                 "macos": null,

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

@@ -891,6 +891,7 @@ impl McpProcess {
             id: RequestId::Integer(request_id),
             method: method.to_string(),
             params,
+            trace: None,
         });
         self.send_jsonrpc_message(message).await?;
         Ok(request_id)

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

@@ -34,6 +34,7 @@ fn preset_to_info(preset: &ModelPreset, priority: i32) -> ModelInfo {
         default_reasoning_summary: ReasoningSummary::Auto,
         support_verbosity: false,
         default_verbosity: None,
+        availability_nux: None,
         apply_patch_tool_type: None,
         truncation_policy: TruncationPolicyConfig::bytes(10_000),
         supports_parallel_tool_calls: false,

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

@@ -84,6 +84,7 @@ pub fn create_fake_rollout_with_source(
         model_provider: model_provider.map(str::to_string),
         base_instructions: None,
         dynamic_tools: None,
+        memory_mode: None,
     };
     let payload = serde_json::to_value(SessionMetaLine {
         meta,
@@ -165,6 +166,7 @@ pub fn create_fake_rollout_with_text_elements(
         model_provider: model_provider.map(str::to_string),
         base_instructions: None,
         dynamic_tools: None,
+        memory_mode: None,
     };
     let payload = serde_json::to_value(SessionMetaLine {
         meta,

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

@@ -36,7 +36,7 @@ use std::path::Path;
 use tempfile::TempDir;
 use tokio::time::timeout;
 
-const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(45);
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn test_codex_jsonrpc_conversation_flow() -> Result<()> {
@@ -337,6 +337,7 @@ async fn test_send_user_turn_changes_approval_policy_behavior() -> Result<()> {
             model: "mock-model".to_string(),
             effort: Some(ReasoningEffort::Medium),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: None,
         })
         .await?;
@@ -453,6 +454,7 @@ async fn test_send_user_turn_updates_sandbox_and_cwd_between_turns() -> Result<(
             model: model.clone(),
             effort: Some(ReasoningEffort::Medium),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: None,
         })
         .await?;
@@ -481,6 +483,7 @@ async fn test_send_user_turn_updates_sandbox_and_cwd_between_turns() -> Result<(
             model: model.clone(),
             effort: Some(ReasoningEffort::Medium),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: None,
         })
         .await?;

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

@@ -92,6 +92,7 @@ async fn send_user_turn_accepts_output_schema_v1() -> Result<()> {
             model: "mock-model".to_string(),
             effort: Some(ReasoningEffort::Medium),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: Some(output_schema.clone()),
         })
         .await?;
@@ -184,6 +185,7 @@ async fn send_user_turn_rejects_oversized_input_v1() -> Result<()> {
             model: "mock-model".to_string(),
             effort: Some(ReasoningEffort::Low),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: None,
         })
         .await?;
@@ -273,6 +275,7 @@ async fn send_user_turn_output_schema_is_per_turn_v1() -> Result<()> {
             model: "mock-model".to_string(),
             effort: Some(ReasoningEffort::Medium),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: Some(output_schema.clone()),
         })
         .await?;
@@ -321,6 +324,7 @@ async fn send_user_turn_output_schema_is_per_turn_v1() -> Result<()> {
             model: "mock-model".to_string(),
             effort: Some(ReasoningEffort::Medium),
             summary: ReasoningSummary::Auto,
+            service_tier: None,
             output_schema: None,
         })
         .await?;

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

@@ -620,12 +620,15 @@ fn append_rollout_turn_context(path: &Path, timestamp: &str, model: &str) -> std
         item: RolloutItem::TurnContext(TurnContextItem {
             turn_id: None,
             cwd: PathBuf::from("/"),
+            current_date: None,
+            timezone: None,
             approval_policy: AskForApproval::Never,
             sandbox_policy: SandboxPolicy::DangerFullAccess,
             network: None,
             model: model.to_string(),
             personality: None,
             collaboration_mode: None,
+            realtime_active: Some(false),
             effort: None,
             summary: ReasoningSummary::Auto,
             user_instructions: None,

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

@@ -131,6 +131,7 @@ async fn logout_account_removes_auth_and_notifies() -> Result<()> {
         payload.auth_mode.is_none(),
         "auth_method should be None after logout"
     );
+    assert_eq!(payload.plan_type, None);
 
     assert!(
         !codex_home.path().join("auth.json").exists(),
@@ -201,6 +202,7 @@ async fn set_auth_token_updates_account_and_notifies() -> Result<()> {
         bail!("unexpected notification: {parsed:?}");
     };
     assert_eq!(payload.auth_mode, Some(AuthMode::ChatgptAuthTokens));
+    assert_eq!(payload.plan_type, Some(AccountPlanType::Pro));
 
     let get_id = mcp
         .send_get_account_request(GetAccountParams {
@@ -843,6 +845,7 @@ async fn login_account_api_key_succeeds_and_notifies() -> Result<()> {
         bail!("unexpected notification: {parsed:?}");
     };
     pretty_assertions::assert_eq!(payload.auth_mode, Some(AuthMode::ApiKey));
+    pretty_assertions::assert_eq!(payload.plan_type, None);
 
     assert!(codex_home.path().join("auth.json").exists());
     Ok(())
@@ -1227,3 +1230,45 @@ async fn get_account_with_chatgpt() -> Result<()> {
     assert_eq!(received, expected);
     Ok(())
 }
+
+#[tokio::test]
+async fn get_account_with_chatgpt_missing_plan_claim_returns_unknown() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    create_config_toml(
+        codex_home.path(),
+        CreateConfigTomlParams {
+            requires_openai_auth: Some(true),
+            ..Default::default()
+        },
+    )?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("access-chatgpt").email("user@example.com"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    let mut mcp = McpProcess::new_with_env(codex_home.path(), &[("OPENAI_API_KEY", None)]).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let params = GetAccountParams {
+        refresh_token: false,
+    };
+    let request_id = mcp.send_get_account_request(params).await?;
+
+    let resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let received: GetAccountResponse = to_response(resp)?;
+
+    let expected = GetAccountResponse {
+        account: Some(Account::Chatgpt {
+            email: "user@example.com".to_string(),
+            plan_type: AccountPlanType::Unknown,
+        }),
+        requires_openai_auth: true,
+    };
+    assert_eq!(received, expected);
+    Ok(())
+}

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

@@ -30,7 +30,7 @@ async fn app_server_default_analytics_disabled_without_flag() -> Result<()> {
     let provider = codex_core::otel_init::build_provider(
         &config,
         SERVICE_VERSION,
-        Some("codex_app_server"),
+        Some("codex-app-server"),
         false,
     )
     .map_err(|err| anyhow::anyhow!(err.to_string()))?;
@@ -55,7 +55,7 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
     let provider = codex_core::otel_init::build_provider(
         &config,
         SERVICE_VERSION,
-        Some("codex_app_server"),
+        Some("codex-app-server"),
         true,
     )
     .map_err(|err| anyhow::anyhow!(err.to_string()))?;

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

@@ -428,7 +428,7 @@ async fn list_apps_emits_updates_and_returns_after_both_lists_load() -> Result<(
 }
 
 #[tokio::test]
-async fn list_apps_returns_connectors_with_accessible_flags() -> Result<()> {
+async fn list_apps_waits_for_accessible_data_before_emitting_directory_updates() -> Result<()> {
     let connectors = vec![
         AppInfo {
             id: "alpha".to_string(),
@@ -475,7 +475,7 @@ async fn list_apps_returns_connectors_with_accessible_flags() -> Result<()> {
         codex_home.path(),
         ChatGptAuthFixture::new("chatgpt-token")
             .account_id("account-123")
-            .chatgpt_user_id("user-123")
+            .chatgpt_user_id("user-directory-first")
             .chatgpt_account_id("account-123"),
         AuthCredentialsStoreMode::File,
     )?;
@@ -492,60 +492,14 @@ async fn list_apps_returns_connectors_with_accessible_flags() -> Result<()> {
         })
         .await?;
 
-    let expected_directory_first = vec![
-        AppInfo {
-            id: "alpha".to_string(),
-            name: "Alpha".to_string(),
-            description: Some("Alpha connector".to_string()),
-            logo_url: Some("https://example.com/alpha.png".to_string()),
-            logo_url_dark: None,
-            distribution_channel: None,
-            branding: None,
-            app_metadata: None,
-            labels: None,
-            install_url: Some("https://chatgpt.com/apps/alpha/alpha".to_string()),
-            is_accessible: false,
-            is_enabled: true,
-        },
-        AppInfo {
-            id: "beta".to_string(),
-            name: "beta".to_string(),
-            description: None,
-            logo_url: None,
-            logo_url_dark: None,
-            distribution_channel: None,
-            branding: None,
-            app_metadata: None,
-            labels: None,
-            install_url: Some("https://chatgpt.com/apps/beta/beta".to_string()),
-            is_accessible: false,
-            is_enabled: true,
-        },
-    ];
-    let expected_accessible_first = vec![AppInfo {
-        id: "beta".to_string(),
-        name: "Beta App".to_string(),
-        description: None,
-        logo_url: None,
-        logo_url_dark: None,
-        distribution_channel: None,
-        branding: None,
-        app_metadata: None,
-        labels: None,
-        install_url: Some("https://chatgpt.com/apps/beta-app/beta".to_string()),
-        is_accessible: true,
-        is_enabled: true,
-    }];
-
-    let first_update = read_app_list_updated_notification(&mut mcp).await?;
-    // app/list emits an update after whichever async load finishes first. Even with
-    // a tools delay in this test, the accessible-tools path can return first if the
-    // process-global Codex Apps tools cache is warm from another test.
+    let maybe_update = timeout(
+        Duration::from_millis(150),
+        read_app_list_updated_notification(&mut mcp),
+    )
+    .await;
     assert!(
-        first_update.data == expected_directory_first
-            || first_update.data == expected_accessible_first,
-        "unexpected first app/list update: {:#?}",
-        first_update.data
+        maybe_update.is_err(),
+        "unexpected directory-only app/list update before accessible apps loaded"
     );
 
     let expected = vec![
@@ -579,8 +533,96 @@ async fn list_apps_returns_connectors_with_accessible_flags() -> Result<()> {
         },
     ];
 
-    let second_update = read_app_list_updated_notification(&mut mcp).await?;
-    assert_eq!(second_update.data, expected);
+    let update = read_app_list_updated_notification(&mut mcp).await?;
+    assert_eq!(update.data, expected);
+
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let AppsListResponse { data, next_cursor } = to_response(response)?;
+    assert_eq!(data, expected);
+    assert!(next_cursor.is_none());
+
+    server_handle.abort();
+    Ok(())
+}
+
+#[tokio::test]
+async fn list_apps_does_not_emit_empty_interim_updates() -> Result<()> {
+    let connectors = vec![AppInfo {
+        id: "alpha".to_string(),
+        name: "Alpha".to_string(),
+        description: Some("Alpha connector".to_string()),
+        logo_url: None,
+        logo_url_dark: None,
+        distribution_channel: None,
+        branding: None,
+        app_metadata: None,
+        labels: None,
+        install_url: None,
+        is_accessible: false,
+        is_enabled: true,
+    }];
+    let (server_url, server_handle) = start_apps_server_with_delays(
+        connectors.clone(),
+        Vec::new(),
+        Duration::from_millis(300),
+        Duration::ZERO,
+    )
+    .await?;
+
+    let codex_home = TempDir::new()?;
+    write_connectors_config(codex_home.path(), &server_url)?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("chatgpt-token")
+            .account_id("account-123")
+            .chatgpt_user_id("user-empty-interim")
+            .chatgpt_account_id("account-123"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+
+    let request_id = mcp
+        .send_apps_list_request(AppsListParams {
+            limit: None,
+            cursor: None,
+            thread_id: None,
+            force_refetch: false,
+        })
+        .await?;
+
+    let maybe_update = timeout(
+        Duration::from_millis(150),
+        read_app_list_updated_notification(&mut mcp),
+    )
+    .await;
+    assert!(
+        maybe_update.is_err(),
+        "unexpected empty interim app/list update"
+    );
+
+    let expected = vec![AppInfo {
+        id: "alpha".to_string(),
+        name: "Alpha".to_string(),
+        description: Some("Alpha connector".to_string()),
+        logo_url: None,
+        logo_url_dark: None,
+        distribution_channel: None,
+        branding: None,
+        app_metadata: None,
+        labels: None,
+        install_url: Some("https://chatgpt.com/apps/alpha/alpha".to_string()),
+        is_accessible: false,
+        is_enabled: true,
+    }];
+
+    let update = read_app_list_updated_notification(&mut mcp).await?;
+    assert_eq!(update.data, expected);
 
     let response: JSONRPCResponse = timeout(
         DEFAULT_TIMEOUT,
@@ -995,6 +1037,20 @@ async fn list_apps_force_refetch_patches_updates_from_cached_snapshots() -> Resu
     assert_eq!(
         first_update.data,
         vec![
+            AppInfo {
+                id: "beta".to_string(),
+                name: "Beta App".to_string(),
+                description: Some("Beta v1".to_string()),
+                logo_url: None,
+                logo_url_dark: None,
+                distribution_channel: None,
+                branding: None,
+                app_metadata: None,
+                labels: None,
+                install_url: Some("https://chatgpt.com/apps/beta-app/beta".to_string()),
+                is_accessible: true,
+                is_enabled: true,
+            },
             AppInfo {
                 id: "alpha".to_string(),
                 name: "Alpha".to_string(),
@@ -1009,23 +1065,19 @@ async fn list_apps_force_refetch_patches_updates_from_cached_snapshots() -> Resu
                 is_accessible: false,
                 is_enabled: true,
             },
-            AppInfo {
-                id: "beta".to_string(),
-                name: "Beta App".to_string(),
-                description: Some("Beta v1".to_string()),
-                logo_url: None,
-                logo_url_dark: None,
-                distribution_channel: None,
-                branding: None,
-                app_metadata: None,
-                labels: None,
-                install_url: Some("https://chatgpt.com/apps/beta-app/beta".to_string()),
-                is_accessible: false,
-                is_enabled: true,
-            },
         ]
     );
 
+    let maybe_second_update = timeout(
+        Duration::from_millis(150),
+        read_app_list_updated_notification(&mut mcp),
+    )
+    .await;
+    assert!(
+        maybe_second_update.is_err(),
+        "unexpected inaccessible-only app/list update during force refetch"
+    );
+
     let expected_final = vec![AppInfo {
         id: "alpha".to_string(),
         name: "Alpha".to_string(),

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

@@ -174,6 +174,7 @@ pub(super) async fn send_request(
         id: RequestId::Integer(id),
         method: method.to_string(),
         params,
+        trace: None,
     });
     send_jsonrpc(stream, message).await
 }

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

@@ -1,16 +1,24 @@
 use anyhow::Result;
 use app_test_support::McpProcess;
+use app_test_support::create_final_assistant_message_sse_response;
 use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::to_response;
 use codex_app_server_protocol::ClientInfo;
 use codex_app_server_protocol::InitializeCapabilities;
 use codex_app_server_protocol::InitializeResponse;
 use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::TurnStartResponse;
+use codex_app_server_protocol::UserInput as V2UserInput;
+use core_test_support::fs_wait;
 use pretty_assertions::assert_eq;
+use serde_json::Value;
 use std::path::Path;
+use std::time::Duration;
 use tempfile::TempDir;
 use tokio::time::timeout;
 
@@ -178,11 +186,103 @@ async fn initialize_opt_out_notification_methods_filters_notifications() -> Resu
     Ok(())
 }
 
+#[tokio::test]
+async fn turn_start_notify_payload_includes_initialize_client_name() -> Result<()> {
+    let responses = vec![create_final_assistant_message_sse_response("Done")?];
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    let notify_script = codex_home.path().join("notify.py");
+    std::fs::write(
+        &notify_script,
+        r#"from pathlib import Path
+import sys
+
+payload_path = Path(__file__).with_name("notify.json")
+tmp_path = payload_path.with_suffix(".json.tmp")
+tmp_path.write_text(sys.argv[-1], encoding="utf-8")
+tmp_path.replace(payload_path)
+"#,
+    )?;
+    let notify_file = codex_home.path().join("notify.json");
+    let notify_script = notify_script
+        .to_str()
+        .expect("notify script path should be valid UTF-8");
+    create_config_toml_with_extra(
+        codex_home.path(),
+        &server.uri(),
+        "never",
+        &format!(
+            "notify = [\"python3\", {}]",
+            toml_basic_string(notify_script)
+        ),
+    )?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.initialize_with_client_info(ClientInfo {
+            name: "xcode".to_string(),
+            title: Some("Xcode".to_string()),
+            version: "1.0.0".to_string(),
+        }),
+    )
+    .await??;
+
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams::default())
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response(thread_resp)?;
+
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![V2UserInput::Text {
+                text: "Hello".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    let turn_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+    let _: TurnStartResponse = to_response(turn_resp)?;
+
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    fs_wait::wait_for_path_exists(&notify_file, Duration::from_secs(5)).await?;
+    let payload_raw = tokio::fs::read_to_string(&notify_file).await?;
+    let payload: Value = serde_json::from_str(&payload_raw)?;
+    assert_eq!(payload["client"], "xcode");
+
+    Ok(())
+}
+
 // Helper to create a config.toml pointing at the mock model server.
 fn create_config_toml(
     codex_home: &Path,
     server_uri: &str,
     approval_policy: &str,
+) -> std::io::Result<()> {
+    create_config_toml_with_extra(codex_home, server_uri, approval_policy, "")
+}
+
+fn create_config_toml_with_extra(
+    codex_home: &Path,
+    server_uri: &str,
+    approval_policy: &str,
+    extra: &str,
 ) -> std::io::Result<()> {
     let config_toml = codex_home.join("config.toml");
     std::fs::write(
@@ -195,6 +295,8 @@ sandbox_mode = "read-only"
 
 model_provider = "mock_provider"
 
+{extra}
+
 [model_providers.mock_provider]
 name = "Mock provider for test"
 base_url = "{server_uri}/v1"
@@ -205,3 +307,7 @@ stream_max_retries = 0
         ),
     )
 }
+
+fn toml_basic_string(value: &str) -> String {
+    format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\""))
+}

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

@@ -9,6 +9,7 @@ use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::Model;
 use codex_app_server_protocol::ModelListParams;
 use codex_app_server_protocol::ModelListResponse;
+use codex_app_server_protocol::ModelUpgradeInfo;
 use codex_app_server_protocol::ReasoningEffortOption;
 use codex_app_server_protocol::RequestId;
 use codex_protocol::openai_models::ModelPreset;
@@ -24,6 +25,13 @@ fn model_from_preset(preset: &ModelPreset) -> Model {
         id: preset.id.clone(),
         model: preset.model.clone(),
         upgrade: preset.upgrade.as_ref().map(|upgrade| upgrade.id.clone()),
+        upgrade_info: preset.upgrade.as_ref().map(|upgrade| ModelUpgradeInfo {
+            model: upgrade.id.clone(),
+            upgrade_copy: upgrade.upgrade_copy.clone(),
+            model_link: upgrade.model_link.clone(),
+            migration_markdown: upgrade.migration_markdown.clone(),
+        }),
+        availability_nux: preset.availability_nux.clone().map(Into::into),
         display_name: preset.display_name.clone(),
         description: preset.description.clone(),
         hidden: !preset.show_in_picker,

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

@@ -5,6 +5,7 @@ use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::to_response;
 use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::LoginApiKeyParams;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
 use codex_app_server_protocol::ThreadRealtimeAppendAudioResponse;
@@ -42,20 +43,17 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
 
     let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
     let realtime_server = start_websocket_server(vec![vec![
-        vec![json!({
-            "type": "session.created",
-            "session": { "id": "sess_backend" }
-        })],
         vec![json!({
             "type": "session.updated",
-            "session": { "backend_prompt": "backend prompt" }
+            "session": { "id": "sess_backend", "instructions": "backend prompt" }
         })],
+        vec![],
         vec![
             json!({
-                "type": "response.output_audio.delta",
+                "type": "conversation.output_audio.delta",
                 "delta": "AQID",
                 "sample_rate": 24_000,
-                "num_channels": 1,
+                "channels": 1,
                 "samples_per_channel": 512
             }),
             json!({
@@ -84,6 +82,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
 
     let mut mcp = McpProcess::new(codex_home.path()).await?;
     mcp.initialize().await?;
+    login_with_api_key(&mut mcp, "sk-test-key").await?;
 
     let thread_start_request_id = mcp
         .send_thread_start_request(ThreadStartParams::default())
@@ -182,7 +181,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
     assert_eq!(connection.len(), 3);
     assert_eq!(
         connection[0].body_json()["type"].as_str(),
-        Some("session.create")
+        Some("session.update")
     );
     let mut request_types = [
         connection[1].body_json()["type"]
@@ -199,7 +198,7 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
         request_types,
         [
             "conversation.item.create".to_string(),
-            "response.input_audio.delta".to_string(),
+            "input_audio_buffer.append".to_string(),
         ]
     );
 
@@ -214,8 +213,8 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
     let responses_server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
     let realtime_server = start_websocket_server(vec![vec![
         vec![json!({
-            "type": "session.created",
-            "session": { "id": "sess_backend" }
+            "type": "session.updated",
+            "session": { "id": "sess_backend", "instructions": "backend prompt" }
         })],
         vec![],
     ]])
@@ -231,6 +230,7 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
 
     let mut mcp = McpProcess::new(codex_home.path()).await?;
     mcp.initialize().await?;
+    login_with_api_key(&mut mcp, "sk-test-key").await?;
 
     let thread_start_request_id = mcp
         .send_thread_start_request(ThreadStartParams::default())
@@ -349,6 +349,22 @@ async fn read_notification<T: DeserializeOwned>(mcp: &mut McpProcess, method: &s
     Ok(serde_json::from_value(params)?)
 }
 
+async fn login_with_api_key(mcp: &mut McpProcess, api_key: &str) -> Result<()> {
+    let request_id = mcp
+        .send_login_api_key_request(LoginApiKeyParams {
+            api_key: api_key.to_string(),
+        })
+        .await?;
+
+    timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+
+    Ok(())
+}
+
 fn create_config_toml(
     codex_home: &Path,
     responses_server_uri: &str,

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

@@ -4,9 +4,11 @@ use app_test_support::create_final_assistant_message_sse_response;
 use app_test_support::create_mock_responses_server_sequence;
 use app_test_support::create_request_user_input_sse_response;
 use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCMessage;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ServerRequest;
+use codex_app_server_protocol::ServerRequestResolvedNotification;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::TurnStartParams;
@@ -86,6 +88,7 @@ async fn request_user_input_round_trip() -> Result<()> {
     assert_eq!(params.turn_id, turn.id);
     assert_eq!(params.item_id, "call1");
     assert_eq!(params.questions.len(), 1);
+    let resolved_request_id = request_id.clone();
 
     mcp.send_response(
         request_id,
@@ -96,17 +99,31 @@ async fn request_user_input_round_trip() -> Result<()> {
         }),
     )
     .await?;
-
-    timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("codex/event/task_complete"),
-    )
-    .await??;
-    timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("turn/completed"),
-    )
-    .await??;
+    let mut saw_resolved = false;
+    loop {
+        let message = timeout(DEFAULT_READ_TIMEOUT, mcp.read_next_message()).await??;
+        let JSONRPCMessage::Notification(notification) = message else {
+            continue;
+        };
+        match notification.method.as_str() {
+            "serverRequest/resolved" => {
+                let resolved: ServerRequestResolvedNotification = serde_json::from_value(
+                    notification
+                        .params
+                        .clone()
+                        .expect("serverRequest/resolved params"),
+                )?;
+                assert_eq!(resolved.thread_id, thread.id);
+                assert_eq!(resolved.request_id, resolved_request_id);
+                saw_resolved = true;
+            }
+            "turn/completed" => {
+                assert!(saw_resolved, "serverRequest/resolved should arrive first");
+                break;
+            }
+            _ => {}
+        }
+    }
 
     Ok(())
 }

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

@@ -8,6 +8,7 @@ use app_test_support::to_response;
 use codex_app_server_protocol::ItemCompletedNotification;
 use codex_app_server_protocol::ItemStartedNotification;
 use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCMessage;
 use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
@@ -19,9 +20,12 @@ use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::ThreadStartedNotification;
+use codex_app_server_protocol::ThreadStatusChangedNotification;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStatus;
 use codex_app_server_protocol::UserInput as V2UserInput;
+use pretty_assertions::assert_eq;
 use serde_json::json;
 use tempfile::TempDir;
 use tokio::time::timeout;
@@ -301,6 +305,31 @@ async fn review_start_with_detached_delivery_returns_new_thread_id() -> Result<(
         "detached review should run on a different thread"
     );
 
+    let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT;
+    let notification = loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let message = timeout(remaining, mcp.read_next_message()).await??;
+        let JSONRPCMessage::Notification(notification) = message else {
+            continue;
+        };
+        if notification.method == "thread/status/changed" {
+            let status_changed: ThreadStatusChangedNotification =
+                serde_json::from_value(notification.params.expect("params must be present"))?;
+            if status_changed.thread_id == review_thread_id {
+                anyhow::bail!(
+                    "detached review threads should be introduced without a preceding thread/status/changed"
+                );
+            }
+            continue;
+        }
+        if notification.method == "thread/started" {
+            break notification;
+        }
+    };
+    let started: ThreadStartedNotification =
+        serde_json::from_value(notification.params.expect("params must be present"))?;
+    assert_eq!(started.thread.id, review_thread_id);
+
     Ok(())
 }
 
@@ -389,6 +418,11 @@ async fn start_default_thread(mcp: &mut McpProcess) -> Result<String> {
     )
     .await??;
     let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("thread/started"),
+    )
+    .await??;
     Ok(thread.id)
 }
 

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

@@ -4,7 +4,7 @@ use app_test_support::create_fake_rollout;
 use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
 use codex_app_server_protocol::JSONRPCError;
-use codex_app_server_protocol::JSONRPCNotification;
+use codex_app_server_protocol::JSONRPCMessage;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::SessionSource;
@@ -15,6 +15,7 @@ use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadStartedNotification;
 use codex_app_server_protocol::ThreadStatus;
+use codex_app_server_protocol::ThreadStatusChangedNotification;
 use codex_app_server_protocol::TurnStatus;
 use codex_app_server_protocol::UserInput;
 use pretty_assertions::assert_eq;
@@ -124,11 +125,27 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
     }
 
     // A corresponding thread/started notification should arrive.
-    let notif: JSONRPCNotification = timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("thread/started"),
-    )
-    .await??;
+    let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT;
+    let notif = loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        let message = timeout(remaining, mcp.read_next_message()).await??;
+        let JSONRPCMessage::Notification(notif) = message else {
+            continue;
+        };
+        if notif.method == "thread/status/changed" {
+            let status_changed: ThreadStatusChangedNotification =
+                serde_json::from_value(notif.params.expect("params must be present"))?;
+            if status_changed.thread_id == thread.id {
+                anyhow::bail!(
+                    "thread/fork should introduce the thread without a preceding thread/status/changed"
+                );
+            }
+            continue;
+        }
+        if notif.method == "thread/started" {
+            break notif;
+        }
+    };
     let started_params = notif.params.clone().expect("params must be present");
     let started_thread_json = started_params
         .get("thread")