Merge f18ba2858e into sapling-pr-archive-bolinfest

.github/workflows/rust-release.yml

@@ -494,9 +494,10 @@ jobs:
             --package codex-responses-api-proxy \
             --package codex-sdk
 
-      - name: Stage macOS and Linux installer script
+      - name: Stage installer scripts
         run: |
           cp scripts/install/install.sh dist/install.sh
+          cp scripts/install/install.ps1 dist/install.ps1
 
       - name: Create GitHub Release
         uses: softprops/action-gh-release@v2

.vscode/settings.json

@@ -1,7 +1,7 @@
 {
     "rust-analyzer.checkOnSave": true,
     "rust-analyzer.check.command": "clippy",
-    "rust-analyzer.check.extraArgs": ["--all-features", "--tests"],
+    "rust-analyzer.check.extraArgs": ["--tests"],
     "rust-analyzer.rustfmt.extraArgs": ["--config", "imports_granularity=Item"],
     "rust-analyzer.cargo.targetDir": "${workspaceFolder}/codex-rs/target/rust-analyzer",
     "[rust]": {

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json

@@ -5,7 +5,7 @@
       "properties": {
         "read": {
           "items": {
-            "$ref": "#/definitions/AbsolutePathBuf"
+            "$ref": "#/definitions/v2/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -14,7 +14,7 @@
         },
         "write": {
           "items": {
-            "$ref": "#/definitions/AbsolutePathBuf"
+            "$ref": "#/definitions/v2/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -1382,7 +1382,7 @@
         "thread_id": {
           "allOf": [
             {
-              "$ref": "#/definitions/ThreadId"
+              "$ref": "#/definitions/v2/ThreadId"
             }
           ],
           "description": "Thread ID of the receiver/new agent."
@@ -1420,7 +1420,7 @@
         "thread_id": {
           "allOf": [
             {
-              "$ref": "#/definitions/ThreadId"
+              "$ref": "#/definitions/v2/ThreadId"
             }
           ],
           "description": "Thread ID of the receiver/new agent."
@@ -4732,7 +4732,7 @@
       "properties": {
         "read": {
           "items": {
-            "$ref": "#/definitions/AbsolutePathBuf"
+            "$ref": "#/definitions/v2/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -4741,7 +4741,7 @@
         },
         "write": {
           "items": {
-            "$ref": "#/definitions/AbsolutePathBuf"
+            "$ref": "#/definitions/v2/AbsolutePathBuf"
           },
           "type": [
             "array",
@@ -7131,7 +7131,7 @@
           "properties": {
             "content": {
               "items": {
-                "$ref": "#/definitions/UserInput"
+                "$ref": "#/definitions/v2/UserInput"
               },
               "type": "array"
             },
@@ -7169,7 +7169,7 @@
             "phase": {
               "anyOf": [
                 {
-                  "$ref": "#/definitions/MessagePhase"
+                  "$ref": "#/definitions/v2/MessagePhase"
                 },
                 {
                   "type": "null"
@@ -7254,7 +7254,7 @@
         {
           "properties": {
             "action": {
-              "$ref": "#/definitions/WebSearchAction"
+              "$ref": "#/definitions/v2/WebSearchAction"
             },
             "id": {
               "type": "string"

codex-rs/app-server-protocol/src/bin/export.rs

@@ -14,9 +14,21 @@ struct Args {
     /// Optional Prettier executable path to format generated TypeScript files
     #[arg(short = 'p', long = "prettier", value_name = "PRETTIER_BIN")]
     prettier: Option<PathBuf>,
+
+    /// Include experimental API methods and fields in generated output.
+    #[arg(long = "experimental")]
+    experimental: bool,
 }
 
 fn main() -> Result<()> {
     let args = Args::parse();
-    codex_app_server_protocol::generate_types(&args.out_dir, args.prettier.as_deref())
+    codex_app_server_protocol::generate_ts_with_options(
+        &args.out_dir,
+        args.prettier.as_deref(),
+        codex_app_server_protocol::GenerateTsOptions {
+            experimental_api: args.experimental,
+            ..codex_app_server_protocol::GenerateTsOptions::default()
+        },
+    )?;
+    codex_app_server_protocol::generate_json_with_experimental(&args.out_dir, args.experimental)
 }

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

@@ -37,6 +37,14 @@ use ts_rs::TS;
 const HEADER: &str = "// GENERATED CODE! DO NOT MODIFY BY HAND!\n\n";
 const IGNORED_DEFINITIONS: &[&str] = &["Option<()>"];
 const JSON_V1_ALLOWLIST: &[&str] = &["InitializeParams", "InitializeResponse"];
+const SPECIAL_DEFINITIONS: &[&str] = &[
+    "ClientNotification",
+    "ClientRequest",
+    "EventMsg",
+    "ServerNotification",
+    "ServerRequest",
+];
+const FLAT_V2_SHARED_DEFINITIONS: &[&str] = &["ClientRequest", "EventMsg", "ServerNotification"];
 const V1_CLIENT_REQUEST_METHODS: &[&str] = &[
     "newConversation",
     "getConversationSummary",
@@ -223,6 +231,11 @@ pub fn generate_json_with_experimental(out_dir: &Path, experimental_api: bool) -
         out_dir.join("codex_app_server_protocol.schemas.json"),
         &bundle,
     )?;
+    let flat_v2_bundle = build_flat_v2_schema(&bundle)?;
+    write_pretty_json(
+        out_dir.join("codex_app_server_protocol.v2.schemas.json"),
+        &flat_v2_bundle,
+    )?;
 
     if !experimental_api {
         filter_experimental_json_files(out_dir)?;
@@ -870,14 +883,6 @@ impl Depth {
 }
 
 fn build_schema_bundle(schemas: Vec<GeneratedSchema>) -> Result<Value> {
-    const SPECIAL_DEFINITIONS: &[&str] = &[
-        "ClientNotification",
-        "ClientRequest",
-        "EventMsg",
-        "ServerNotification",
-        "ServerRequest",
-    ];
-
     let namespaced_types = collect_namespaced_types(&schemas);
     let mut definitions = Map::new();
 
@@ -895,6 +900,8 @@ fn build_schema_bundle(schemas: Vec<GeneratedSchema>) -> Result<Value> {
 
         if let Some(ref ns) = namespace {
             rewrite_refs_to_namespace(&mut value, ns);
+        } else {
+            rewrite_refs_to_known_namespaces(&mut value, &namespaced_types);
         }
 
         let mut forced_namespace_refs: Vec<(String, String)> = Vec::new();
@@ -958,6 +965,210 @@ fn build_schema_bundle(schemas: Vec<GeneratedSchema>) -> Result<Value> {
     Ok(Value::Object(root))
 }
 
+/// Build a datamodel-code-generator-friendly v2 bundle from the mixed export.
+///
+/// The full bundle keeps v2 schemas nested under `definitions.v2`, plus a few
+/// shared root definitions like `ClientRequest`, `EventMsg`, and
+/// `ServerNotification`. Python codegen only walks one definitions map level, so
+/// a direct feed would treat `v2` itself as a schema and miss unreferenced v2
+/// leaves. This helper flattens all v2 definitions to the root definitions map,
+/// then pulls in the shared root schemas and any non-v2 transitive deps they
+/// still reference. Keep the shared root unions intact here: some valid
+/// request/notification/event variants are inline or only reference shared root
+/// helpers, so filtering them by the presence of a `#/definitions/v2/` ref
+/// would silently drop real API surface from the flat bundle.
+fn build_flat_v2_schema(bundle: &Value) -> Result<Value> {
+    let Value::Object(root) = bundle else {
+        return Err(anyhow!("expected bundle root to be an object"));
+    };
+    let definitions = root
+        .get("definitions")
+        .and_then(Value::as_object)
+        .ok_or_else(|| anyhow!("expected bundle definitions map"))?;
+    let v2_definitions = definitions
+        .get("v2")
+        .and_then(Value::as_object)
+        .ok_or_else(|| anyhow!("expected v2 namespace in bundle definitions"))?;
+
+    let mut flat_root = root.clone();
+    let title = root
+        .get("title")
+        .and_then(Value::as_str)
+        .unwrap_or("CodexAppServerProtocol");
+    let mut flat_definitions = v2_definitions.clone();
+    let mut shared_definitions = Map::new();
+    let mut non_v2_refs = HashSet::new();
+
+    for shared in FLAT_V2_SHARED_DEFINITIONS {
+        let Some(shared_schema) = definitions.get(*shared) else {
+            continue;
+        };
+        let shared_schema = shared_schema.clone();
+        non_v2_refs.extend(collect_non_v2_refs(&shared_schema));
+        shared_definitions.insert((*shared).to_string(), shared_schema);
+    }
+
+    for name in collect_definition_dependencies(definitions, non_v2_refs) {
+        if name == "v2" || flat_definitions.contains_key(&name) {
+            continue;
+        }
+        if let Some(schema) = definitions.get(&name) {
+            flat_definitions.insert(name, schema.clone());
+        }
+    }
+
+    flat_definitions.extend(shared_definitions);
+    flat_root.insert("title".to_string(), Value::String(format!("{title}V2")));
+    flat_root.insert("definitions".to_string(), Value::Object(flat_definitions));
+    let mut flat_bundle = Value::Object(flat_root);
+    rewrite_ref_prefix(&mut flat_bundle, "#/definitions/v2/", "#/definitions/");
+    ensure_no_ref_prefix(&flat_bundle, "#/definitions/v2/", "flat v2")?;
+    ensure_referenced_definitions_present(&flat_bundle, "flat v2")?;
+    Ok(flat_bundle)
+}
+
+fn collect_non_v2_refs(value: &Value) -> HashSet<String> {
+    let mut refs = HashSet::new();
+    collect_non_v2_refs_inner(value, &mut refs);
+    refs
+}
+
+fn collect_non_v2_refs_inner(value: &Value, refs: &mut HashSet<String>) {
+    match value {
+        Value::Object(obj) => {
+            if let Some(Value::String(reference)) = obj.get("$ref")
+                && let Some(name) = reference.strip_prefix("#/definitions/")
+                && !reference.starts_with("#/definitions/v2/")
+            {
+                refs.insert(name.to_string());
+            }
+            for child in obj.values() {
+                collect_non_v2_refs_inner(child, refs);
+            }
+        }
+        Value::Array(items) => {
+            for child in items {
+                collect_non_v2_refs_inner(child, refs);
+            }
+        }
+        _ => {}
+    }
+}
+
+fn collect_definition_dependencies(
+    definitions: &Map<String, Value>,
+    names: HashSet<String>,
+) -> HashSet<String> {
+    let mut seen = HashSet::new();
+    let mut to_process: Vec<String> = names.into_iter().collect();
+    while let Some(name) = to_process.pop() {
+        if !seen.insert(name.clone()) {
+            continue;
+        }
+        let Some(schema) = definitions.get(&name) else {
+            continue;
+        };
+        for dep in collect_non_v2_refs(schema) {
+            if !seen.contains(&dep) {
+                to_process.push(dep);
+            }
+        }
+    }
+    seen
+}
+
+fn rewrite_ref_prefix(value: &mut Value, prefix: &str, replacement: &str) {
+    match value {
+        Value::Object(obj) => {
+            if let Some(Value::String(reference)) = obj.get_mut("$ref") {
+                *reference = reference.replace(prefix, replacement);
+            }
+            for child in obj.values_mut() {
+                rewrite_ref_prefix(child, prefix, replacement);
+            }
+        }
+        Value::Array(items) => {
+            for child in items {
+                rewrite_ref_prefix(child, prefix, replacement);
+            }
+        }
+        _ => {}
+    }
+}
+
+fn ensure_no_ref_prefix(value: &Value, prefix: &str, label: &str) -> Result<()> {
+    if let Some(reference) = first_ref_with_prefix(value, prefix) {
+        return Err(anyhow!(
+            "{label} schema still references namespaced definitions; found {reference}"
+        ));
+    }
+    Ok(())
+}
+
+fn first_ref_with_prefix(value: &Value, prefix: &str) -> Option<String> {
+    match value {
+        Value::Object(obj) => {
+            if let Some(Value::String(reference)) = obj.get("$ref")
+                && reference.starts_with(prefix)
+            {
+                return Some(reference.clone());
+            }
+            obj.values()
+                .find_map(|child| first_ref_with_prefix(child, prefix))
+        }
+        Value::Array(items) => items
+            .iter()
+            .find_map(|child| first_ref_with_prefix(child, prefix)),
+        _ => None,
+    }
+}
+
+fn ensure_referenced_definitions_present(schema: &Value, label: &str) -> Result<()> {
+    let definitions = schema
+        .get("definitions")
+        .and_then(Value::as_object)
+        .ok_or_else(|| anyhow!("expected definitions map in {label} schema"))?;
+    let mut missing = HashSet::new();
+    collect_missing_definitions(schema, definitions, &mut missing);
+    if missing.is_empty() {
+        return Ok(());
+    }
+    let mut missing_names: Vec<String> = missing.into_iter().collect();
+    missing_names.sort();
+    Err(anyhow!(
+        "{label} schema missing definitions: {}",
+        missing_names.join(", ")
+    ))
+}
+
+fn collect_missing_definitions(
+    value: &Value,
+    definitions: &Map<String, Value>,
+    missing: &mut HashSet<String>,
+) {
+    match value {
+        Value::Object(obj) => {
+            if let Some(Value::String(reference)) = obj.get("$ref")
+                && let Some(name) = reference.strip_prefix("#/definitions/")
+            {
+                let name = name.split('/').next().unwrap_or(name);
+                if !definitions.contains_key(name) {
+                    missing.insert(name.to_string());
+                }
+            }
+            for child in obj.values() {
+                collect_missing_definitions(child, definitions, missing);
+            }
+        }
+        Value::Array(items) => {
+            for child in items {
+                collect_missing_definitions(child, definitions, missing);
+            }
+        }
+        _ => {}
+    }
+}
+
 fn insert_into_namespace(
     definitions: &mut Map<String, Value>,
     namespace: &str,
@@ -1230,6 +1441,43 @@ fn rewrite_refs_to_namespace(value: &mut Value, ns: &str) {
     }
 }
 
+/// Recursively rewrite bare root definition refs to the namespace that owns the
+/// referenced type in the bundle.
+///
+/// The mixed export contains shared root helper schemas that are intentionally
+/// left outside the `v2` namespace, but some of their extracted child
+/// definitions still contain refs like `#/definitions/ThreadId`. When the real
+/// schema only exists under `#/definitions/v2/ThreadId`, those refs become
+/// dangling and downstream codegen falls back to placeholder `Any` models. This
+/// rewrite keeps the shared helpers at the root while retargeting their refs to
+/// the namespaced definitions that actually exist.
+fn rewrite_refs_to_known_namespaces(value: &mut Value, types: &HashMap<String, String>) {
+    match value {
+        Value::Object(obj) => {
+            if let Some(Value::String(reference)) = obj.get_mut("$ref")
+                && let Some(suffix) = reference.strip_prefix("#/definitions/")
+            {
+                let (name, tail) = suffix
+                    .split_once('/')
+                    .map_or((suffix, None), |(name, tail)| (name, Some(tail)));
+                if let Some(ns) = namespace_for_definition(name, types) {
+                    let tail = tail.map_or(String::new(), |rest| format!("/{rest}"));
+                    *reference = format!("#/definitions/{ns}/{name}{tail}");
+                }
+            }
+            for v in obj.values_mut() {
+                rewrite_refs_to_known_namespaces(v, types);
+            }
+        }
+        Value::Array(items) => {
+            for v in items.iter_mut() {
+                rewrite_refs_to_known_namespaces(v, types);
+            }
+        }
+        _ => {}
+    }
+}
+
 fn collect_namespaced_types(schemas: &[GeneratedSchema]) -> HashMap<String, String> {
     let mut types = HashMap::new();
     for schema in schemas {
@@ -1986,6 +2234,284 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn build_schema_bundle_rewrites_root_helper_refs_to_namespaced_defs() -> Result<()> {
+        let bundle = build_schema_bundle(vec![
+            GeneratedSchema {
+                namespace: None,
+                logical_name: "LegacyEnvelope".to_string(),
+                in_v1_dir: false,
+                value: serde_json::json!({
+                    "title": "LegacyEnvelope",
+                    "type": "object",
+                    "properties": {
+                        "current_thread": { "$ref": "#/definitions/ThreadId" },
+                        "turn_item": { "$ref": "#/definitions/TurnItem" }
+                    },
+                    "definitions": {
+                        "TurnItem": {
+                            "type": "object",
+                            "properties": {
+                                "thread_id": { "$ref": "#/definitions/ThreadId" },
+                                "phase": { "$ref": "#/definitions/MessagePhase" },
+                                "content": {
+                                    "type": "array",
+                                    "items": { "$ref": "#/definitions/UserInput" }
+                                }
+                            }
+                        }
+                    }
+                }),
+            },
+            GeneratedSchema {
+                namespace: Some("v2".to_string()),
+                logical_name: "ThreadId".to_string(),
+                in_v1_dir: false,
+                value: serde_json::json!({
+                    "title": "ThreadId",
+                    "type": "string"
+                }),
+            },
+            GeneratedSchema {
+                namespace: Some("v2".to_string()),
+                logical_name: "MessagePhase".to_string(),
+                in_v1_dir: false,
+                value: serde_json::json!({
+                    "title": "MessagePhase",
+                    "type": "string"
+                }),
+            },
+            GeneratedSchema {
+                namespace: Some("v2".to_string()),
+                logical_name: "UserInput".to_string(),
+                in_v1_dir: false,
+                value: serde_json::json!({
+                    "title": "UserInput",
+                    "type": "string"
+                }),
+            },
+        ])?;
+
+        assert_eq!(
+            bundle["definitions"]["LegacyEnvelope"]["properties"]["current_thread"]["$ref"],
+            serde_json::json!("#/definitions/v2/ThreadId")
+        );
+        assert_eq!(
+            bundle["definitions"]["LegacyEnvelope"]["properties"]["turn_item"]["$ref"],
+            serde_json::json!("#/definitions/TurnItem")
+        );
+        assert_eq!(
+            bundle["definitions"]["TurnItem"]["properties"]["thread_id"]["$ref"],
+            serde_json::json!("#/definitions/v2/ThreadId")
+        );
+        assert_eq!(
+            bundle["definitions"]["TurnItem"]["properties"]["phase"]["$ref"],
+            serde_json::json!("#/definitions/v2/MessagePhase")
+        );
+        assert_eq!(
+            bundle["definitions"]["TurnItem"]["properties"]["content"]["items"]["$ref"],
+            serde_json::json!("#/definitions/v2/UserInput")
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn build_flat_v2_schema_keeps_shared_root_schemas_and_dependencies() -> Result<()> {
+        let bundle = serde_json::json!({
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "title": "CodexAppServerProtocol",
+            "type": "object",
+            "definitions": {
+                "ClientRequest": {
+                    "oneOf": [
+                        {
+                            "title": "StartRequest",
+                            "type": "object",
+                            "properties": {
+                                "params": { "$ref": "#/definitions/v2/ThreadStartParams" },
+                                "shared": { "$ref": "#/definitions/SharedHelper" }
+                            }
+                        },
+                        {
+                            "title": "InitializeRequest",
+                            "type": "object",
+                            "properties": {
+                                "params": { "$ref": "#/definitions/InitializeParams" }
+                            }
+                        },
+                        {
+                            "title": "LogoutRequest",
+                            "type": "object",
+                            "properties": {
+                                "params": { "type": "null" }
+                            }
+                        }
+                    ]
+                },
+                "EventMsg": {
+                    "oneOf": [
+                        { "$ref": "#/definitions/v2/ThreadStartedEventMsg" },
+                        {
+                            "title": "WarningEventMsg",
+                            "type": "object",
+                            "properties": {
+                                "message": { "type": "string" },
+                                "type": {
+                                    "enum": ["warning"],
+                                    "type": "string"
+                                }
+                            },
+                            "required": ["message", "type"]
+                        }
+                    ]
+                },
+                "ServerNotification": {
+                    "oneOf": [
+                        { "$ref": "#/definitions/v2/ThreadStartedNotification" },
+                        {
+                            "title": "ServerRequestResolvedNotification",
+                            "type": "object",
+                            "properties": {
+                                "params": { "$ref": "#/definitions/ServerRequestResolvedNotificationPayload" }
+                            }
+                        }
+                    ]
+                },
+                "SharedHelper": {
+                    "type": "object",
+                    "properties": {
+                        "leaf": { "$ref": "#/definitions/SharedLeaf" }
+                    }
+                },
+                "SharedLeaf": {
+                    "title": "SharedLeaf",
+                    "type": "string"
+                },
+                "InitializeParams": {
+                    "title": "InitializeParams",
+                    "type": "string"
+                },
+                "ServerRequestResolvedNotificationPayload": {
+                    "title": "ServerRequestResolvedNotificationPayload",
+                    "type": "string"
+                },
+                "v2": {
+                    "ThreadStartParams": {
+                        "title": "ThreadStartParams",
+                        "type": "object",
+                        "properties": {
+                            "cwd": { "type": "string" }
+                        }
+                    },
+                    "ThreadStartResponse": {
+                        "title": "ThreadStartResponse",
+                        "type": "object",
+                        "properties": {
+                            "ok": { "type": "boolean" }
+                        }
+                    },
+                    "ThreadStartedEventMsg": {
+                        "title": "ThreadStartedEventMsg",
+                        "type": "object",
+                        "properties": {
+                            "thread_id": { "type": "string" }
+                        }
+                    },
+                    "ThreadStartedNotification": {
+                        "title": "ThreadStartedNotification",
+                        "type": "object",
+                        "properties": {
+                            "thread_id": { "type": "string" }
+                        }
+                    }
+                }
+            }
+        });
+
+        let flat_bundle = build_flat_v2_schema(&bundle)?;
+        let definitions = flat_bundle["definitions"]
+            .as_object()
+            .expect("flat v2 schema should include definitions");
+
+        assert_eq!(
+            flat_bundle["title"],
+            serde_json::json!("CodexAppServerProtocolV2")
+        );
+        assert_eq!(definitions.contains_key("v2"), false);
+        assert_eq!(definitions.contains_key("ThreadStartParams"), true);
+        assert_eq!(definitions.contains_key("ThreadStartResponse"), true);
+        assert_eq!(definitions.contains_key("ThreadStartedEventMsg"), true);
+        assert_eq!(definitions.contains_key("ThreadStartedNotification"), true);
+        assert_eq!(definitions.contains_key("SharedHelper"), true);
+        assert_eq!(definitions.contains_key("SharedLeaf"), true);
+        assert_eq!(definitions.contains_key("InitializeParams"), true);
+        assert_eq!(
+            definitions.contains_key("ServerRequestResolvedNotificationPayload"),
+            true
+        );
+        let client_request_titles: BTreeSet<String> = definitions["ClientRequest"]["oneOf"]
+            .as_array()
+            .expect("ClientRequest should remain a oneOf")
+            .iter()
+            .map(|variant| {
+                variant["title"]
+                    .as_str()
+                    .expect("ClientRequest variant should have a title")
+                    .to_string()
+            })
+            .collect();
+        assert_eq!(
+            client_request_titles,
+            BTreeSet::from([
+                "InitializeRequest".to_string(),
+                "LogoutRequest".to_string(),
+                "StartRequest".to_string(),
+            ])
+        );
+        let event_titles: BTreeSet<String> = definitions["EventMsg"]["oneOf"]
+            .as_array()
+            .expect("EventMsg should remain a oneOf")
+            .iter()
+            .map(|variant| {
+                variant
+                    .get("title")
+                    .and_then(Value::as_str)
+                    .unwrap_or_default()
+                    .to_string()
+            })
+            .collect();
+        assert_eq!(
+            event_titles,
+            BTreeSet::from(["".to_string(), "WarningEventMsg".to_string(),])
+        );
+        let notification_titles: BTreeSet<String> = definitions["ServerNotification"]["oneOf"]
+            .as_array()
+            .expect("ServerNotification should remain a oneOf")
+            .iter()
+            .map(|variant| {
+                variant
+                    .get("title")
+                    .and_then(Value::as_str)
+                    .unwrap_or_default()
+                    .to_string()
+            })
+            .collect();
+        assert_eq!(
+            notification_titles,
+            BTreeSet::from([
+                "".to_string(),
+                "ServerRequestResolvedNotification".to_string(),
+            ])
+        );
+        assert_eq!(
+            first_ref_with_prefix(&flat_bundle, "#/definitions/v2/").is_none(),
+            true
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn experimental_type_fields_ts_filter_handles_interface_shape() -> Result<()> {
         let output_dir = std::env::temp_dir().join(format!("codex_ts_filter_{}", Uuid::now_v7()));
@@ -2114,6 +2640,99 @@ export type Config = { stableField: Keep, unstableField: string | null } & ({ [k
             bundle_json.contains("MockExperimentalMethodResponse"),
             false
         );
+        let flat_v2_bundle_json =
+            fs::read_to_string(output_dir.join("codex_app_server_protocol.v2.schemas.json"))?;
+        assert_eq!(flat_v2_bundle_json.contains("mockExperimentalField"), false);
+        assert_eq!(flat_v2_bundle_json.contains("additionalPermissions"), false);
+        assert_eq!(
+            flat_v2_bundle_json.contains("MockExperimentalMethodParams"),
+            false
+        );
+        assert_eq!(
+            flat_v2_bundle_json.contains("MockExperimentalMethodResponse"),
+            false
+        );
+        assert_eq!(flat_v2_bundle_json.contains("#/definitions/v2/"), false);
+        assert_eq!(
+            flat_v2_bundle_json.contains("\"title\": \"CodexAppServerProtocolV2\""),
+            true
+        );
+        let flat_v2_bundle =
+            read_json_value(&output_dir.join("codex_app_server_protocol.v2.schemas.json"))?;
+        let definitions = flat_v2_bundle["definitions"]
+            .as_object()
+            .expect("flat v2 bundle should include definitions");
+        let client_request_methods: BTreeSet<String> = definitions["ClientRequest"]["oneOf"]
+            .as_array()
+            .expect("flat v2 ClientRequest should remain a oneOf")
+            .iter()
+            .filter_map(|variant| {
+                variant["properties"]["method"]["enum"]
+                    .as_array()
+                    .and_then(|values| values.first())
+                    .and_then(Value::as_str)
+                    .map(str::to_string)
+            })
+            .collect();
+        let missing_client_request_methods: Vec<String> = [
+            "account/logout",
+            "account/rateLimits/read",
+            "config/mcpServer/reload",
+            "configRequirements/read",
+            "fuzzyFileSearch",
+            "initialize",
+        ]
+        .into_iter()
+        .filter(|method| !client_request_methods.contains(*method))
+        .map(str::to_string)
+        .collect();
+        assert_eq!(missing_client_request_methods, Vec::<String>::new());
+        let server_notification_methods: BTreeSet<String> =
+            definitions["ServerNotification"]["oneOf"]
+                .as_array()
+                .expect("flat v2 ServerNotification should remain a oneOf")
+                .iter()
+                .filter_map(|variant| {
+                    variant["properties"]["method"]["enum"]
+                        .as_array()
+                        .and_then(|values| values.first())
+                        .and_then(Value::as_str)
+                        .map(str::to_string)
+                })
+                .collect();
+        let missing_server_notification_methods: Vec<String> = [
+            "fuzzyFileSearch/sessionCompleted",
+            "fuzzyFileSearch/sessionUpdated",
+            "serverRequest/resolved",
+        ]
+        .into_iter()
+        .filter(|method| !server_notification_methods.contains(*method))
+        .map(str::to_string)
+        .collect();
+        assert_eq!(missing_server_notification_methods, Vec::<String>::new());
+        let event_types: BTreeSet<String> = definitions["EventMsg"]["oneOf"]
+            .as_array()
+            .expect("flat v2 EventMsg should remain a oneOf")
+            .iter()
+            .filter_map(|variant| {
+                variant["properties"]["type"]["enum"]
+                    .as_array()
+                    .and_then(|values| values.first())
+                    .and_then(Value::as_str)
+                    .map(str::to_string)
+            })
+            .collect();
+        let missing_event_types: Vec<String> = [
+            "agent_message_delta",
+            "task_complete",
+            "warning",
+            "web_search_begin",
+        ]
+        .into_iter()
+        .filter(|event_type| !event_types.contains(*event_type))
+        .map(str::to_string)
+        .collect();
+        assert_eq!(missing_event_types, Vec::<String>::new());
         assert_eq!(
             output_dir
                 .join("v2")

codex-rs/app-server/Cargo.toml

@@ -32,6 +32,7 @@ codex-protocol = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-feedback = { workspace = true }
 codex-rmcp-client = { workspace = true }
+codex-state = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-json-to-toml = { workspace = true }
 chrono = { workspace = true }

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

@@ -274,6 +274,7 @@ use codex_protocol::protocol::USER_MESSAGE_BEGIN;
 use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
 use codex_protocol::user_input::UserInput as CoreInputItem;
 use codex_rmcp_client::perform_oauth_login_return_url;
+use codex_state::log_db::LogDbLayer;
 use codex_utils_json_to_toml::json_to_toml;
 use std::collections::HashMap;
 use std::collections::HashSet;
@@ -383,6 +384,7 @@ pub(crate) struct CodexMessageProcessor {
     pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
     fuzzy_search_sessions: Arc<Mutex<HashMap<String, FuzzyFileSearchSession>>>,
     feedback: CodexFeedback,
+    log_db: Option<LogDbLayer>,
 }
 
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
@@ -417,6 +419,7 @@ pub(crate) struct CodexMessageProcessorArgs {
     pub(crate) cli_overrides: Vec<(String, TomlValue)>,
     pub(crate) cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
     pub(crate) feedback: CodexFeedback,
+    pub(crate) log_db: Option<LogDbLayer>,
 }
 
 impl CodexMessageProcessor {
@@ -461,6 +464,7 @@ impl CodexMessageProcessor {
             cli_overrides,
             cloud_requirements,
             feedback,
+            log_db,
         } = args;
         Self {
             auth_manager,
@@ -477,6 +481,7 @@ impl CodexMessageProcessor {
             pending_fuzzy_searches: Arc::new(Mutex::new(HashMap::new())),
             fuzzy_search_sessions: Arc::new(Mutex::new(HashMap::new())),
             feedback,
+            log_db,
         }
     }
 
@@ -6944,6 +6949,30 @@ impl CodexMessageProcessor {
 
         let snapshot = self.feedback.snapshot(conversation_id);
         let thread_id = snapshot.thread_id.clone();
+        let sqlite_feedback_logs = if include_logs {
+            if let Some(log_db) = self.log_db.as_ref() {
+                log_db.flush().await;
+            }
+            let state_db_ctx = get_state_db(&self.config, None).await;
+            match (state_db_ctx.as_ref(), conversation_id) {
+                (Some(state_db_ctx), Some(conversation_id)) => {
+                    let thread_id_text = conversation_id.to_string();
+                    match state_db_ctx.query_feedback_logs(&thread_id_text).await {
+                        Ok(logs) if logs.is_empty() => None,
+                        Ok(logs) => Some(logs),
+                        Err(err) => {
+                            warn!(
+                                "failed to query feedback logs from sqlite for thread_id={thread_id_text}: {err}"
+                            );
+                            None
+                        }
+                    }
+                }
+                _ => None,
+            }
+        } else {
+            None
+        };
 
         let validated_rollout_path = if include_logs {
             match conversation_id {
@@ -6967,6 +6996,7 @@ impl CodexMessageProcessor {
                 include_logs,
                 &attachment_paths,
                 Some(session_source),
+                sqlite_feedback_logs,
             )
         })
         .await;

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

@@ -38,16 +38,20 @@ use codex_core::ExecPolicyError;
 use codex_core::check_execpolicy_for_warnings;
 use codex_core::config_loader::ConfigLoadError;
 use codex_core::config_loader::TextRange as CoreTextRange;
+use codex_core::features::Feature;
 use codex_feedback::CodexFeedback;
+use codex_state::log_db;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use toml::Value as TomlValue;
+use tracing::Level;
 use tracing::error;
 use tracing::info;
 use tracing::warn;
 use tracing_subscriber::EnvFilter;
 use tracing_subscriber::Layer;
+use tracing_subscriber::filter::Targets;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::registry::Registry;
 use tracing_subscriber::util::SubscriberInitExt;
@@ -477,6 +481,21 @@ pub async fn run_main_with_transport(
 
     let feedback_layer = feedback.logger_layer();
     let feedback_metadata_layer = feedback.metadata_layer();
+    let log_db = if config.features.enabled(Feature::Sqlite) {
+        codex_state::StateRuntime::init(
+            config.sqlite_home.clone(),
+            config.model_provider_id.clone(),
+            None,
+        )
+        .await
+        .ok()
+        .map(log_db::start)
+    } else {
+        None
+    };
+    let log_db_layer = log_db
+        .clone()
+        .map(|layer| layer.with_filter(Targets::new().with_default(Level::TRACE)));
     let otel_logger_layer = otel.as_ref().and_then(|o| o.logger_layer());
     let otel_tracing_layer = otel.as_ref().and_then(|o| o.tracing_layer());
 
@@ -484,6 +503,7 @@ pub async fn run_main_with_transport(
         .with(stderr_fmt)
         .with(feedback_layer)
         .with(feedback_metadata_layer)
+        .with(log_db_layer)
         .with(otel_logger_layer)
         .with(otel_tracing_layer)
         .try_init();
@@ -562,6 +582,7 @@ pub async fn run_main_with_transport(
             loader_overrides,
             cloud_requirements: cloud_requirements.clone(),
             feedback: feedback.clone(),
+            log_db,
             config_warnings,
         });
         let mut thread_created_rx = processor.thread_created_receiver();

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

@@ -54,6 +54,7 @@ use codex_core::models_manager::collaboration_mode_presets::CollaborationModesCo
 use codex_feedback::CodexFeedback;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::SessionSource;
+use codex_state::log_db::LogDbLayer;
 use futures::FutureExt;
 use tokio::sync::broadcast;
 use tokio::sync::watch;
@@ -154,6 +155,7 @@ pub(crate) struct MessageProcessorArgs {
     pub(crate) loader_overrides: LoaderOverrides,
     pub(crate) cloud_requirements: CloudRequirementsLoader,
     pub(crate) feedback: CodexFeedback,
+    pub(crate) log_db: Option<LogDbLayer>,
     pub(crate) config_warnings: Vec<ConfigWarningNotification>,
 }
 
@@ -169,6 +171,7 @@ impl MessageProcessor {
             loader_overrides,
             cloud_requirements,
             feedback,
+            log_db,
             config_warnings,
         } = args;
         let auth_manager = AuthManager::shared(
@@ -201,6 +204,7 @@ impl MessageProcessor {
             cli_overrides: cli_overrides.clone(),
             cloud_requirements: cloud_requirements.clone(),
             feedback,
+            log_db,
         });
         let config_api = ConfigApi::new(
             config.codex_home.clone(),

codex-rs/codex-api/src/common.rs

@@ -67,8 +67,6 @@ pub enum ResponseEvent {
     Completed {
         response_id: String,
         token_usage: Option<TokenUsage>,
-        /// Whether the client can append more items to a long-running websocket response.
-        can_append: bool,
     },
     OutputTextDelta(String),
     ReasoningSummaryDelta {
@@ -211,20 +209,12 @@ pub struct ResponseCreateWsRequest {
     pub client_metadata: Option<HashMap<String, String>>,
 }
 
-#[derive(Debug, Serialize)]
-pub struct ResponseAppendWsRequest {
-    pub input: Vec<ResponseItem>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub client_metadata: Option<HashMap<String, String>>,
-}
 #[derive(Debug, Serialize)]
 #[serde(tag = "type")]
 #[allow(clippy::large_enum_variant)]
 pub enum ResponsesWsRequest {
     #[serde(rename = "response.create")]
     ResponseCreate(ResponseCreateWsRequest),
-    #[serde(rename = "response.append")]
-    ResponseAppend(ResponseAppendWsRequest),
 }
 
 pub fn create_text_param_for_request(

codex-rs/codex-api/src/lib.rs

@@ -19,7 +19,6 @@ pub use crate::common::MemorySummarizeInput;
 pub use crate::common::MemorySummarizeOutput;
 pub use crate::common::RawMemory;
 pub use crate::common::RawMemoryMetadata;
-pub use crate::common::ResponseAppendWsRequest;
 pub use crate::common::ResponseCreateWsRequest;
 pub use crate::common::ResponseEvent;
 pub use crate::common::ResponseStream;

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

@@ -118,14 +118,6 @@ struct ResponseCompleted {
     usage: Option<ResponseCompletedUsage>,
 }
 
-#[derive(Debug, Deserialize)]
-struct ResponseDone {
-    #[serde(default)]
-    id: Option<String>,
-    #[serde(default)]
-    usage: Option<ResponseCompletedUsage>,
-}
-
 #[derive(Debug, Deserialize)]
 struct ResponseCompletedUsage {
     input_tokens: i64,
@@ -324,7 +316,6 @@ pub fn process_responses_event(
                         return Ok(Some(ResponseEvent::Completed {
                             response_id: resp.id,
                             token_usage: resp.usage.map(Into::into),
-                            can_append: false,
                         }));
                     }
                     Err(err) => {
@@ -335,31 +326,6 @@ pub fn process_responses_event(
                 }
             }
         }
-        "response.done" => {
-            if let Some(resp_val) = event.response {
-                match serde_json::from_value::<ResponseDone>(resp_val) {
-                    Ok(resp) => {
-                        return Ok(Some(ResponseEvent::Completed {
-                            response_id: resp.id.unwrap_or_default(),
-                            token_usage: resp.usage.map(Into::into),
-                            can_append: true,
-                        }));
-                    }
-                    Err(err) => {
-                        let error = format!("failed to parse ResponseCompleted: {err}");
-                        debug!("{error}");
-                        return Err(ResponsesEventError::Api(ApiError::Stream(error)));
-                    }
-                }
-            }
-
-            debug!("response.done missing response payload");
-            return Ok(Some(ResponseEvent::Completed {
-                response_id: String::new(),
-                token_usage: None,
-                can_append: true,
-            }));
-        }
         "response.output_item.added" => {
             if let Some(item_val) = event.item {
                 if let Ok(item) = serde_json::from_value::<ResponseItem>(item_val) {
@@ -639,11 +605,9 @@ mod tests {
             Ok(ResponseEvent::Completed {
                 response_id,
                 token_usage,
-                can_append,
             }) => {
                 assert_eq!(response_id, "resp1");
                 assert!(token_usage.is_none());
-                assert!(!can_append);
             }
             other => panic!("unexpected third event: {other:?}"),
         }
@@ -677,69 +641,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn response_done_emits_incremental_completed() {
-        let done = json!({
-            "type": "response.done",
-            "response": {
-                "usage": {
-                    "input_tokens": 1,
-                    "input_tokens_details": null,
-                    "output_tokens": 2,
-                    "output_tokens_details": null,
-                    "total_tokens": 3
-                }
-            }
-        })
-        .to_string();
-
-        let sse1 = format!("event: response.done\ndata: {done}\n\n");
-
-        let events = collect_events(&[sse1.as_bytes()]).await;
-
-        assert_eq!(events.len(), 1);
-
-        match &events[0] {
-            Ok(ResponseEvent::Completed {
-                response_id,
-                token_usage,
-                can_append,
-            }) => {
-                assert_eq!(response_id, "");
-                assert!(token_usage.is_some());
-                assert!(*can_append);
-            }
-            other => panic!("unexpected event: {other:?}"),
-        }
-    }
-
-    #[tokio::test]
-    async fn response_done_without_payload_emits_completed() {
-        let done = json!({
-            "type": "response.done"
-        })
-        .to_string();
-
-        let sse1 = format!("event: response.done\ndata: {done}\n\n");
-
-        let events = collect_events(&[sse1.as_bytes()]).await;
-
-        assert_eq!(events.len(), 1);
-
-        match &events[0] {
-            Ok(ResponseEvent::Completed {
-                response_id,
-                token_usage,
-                can_append,
-            }) => {
-                assert_eq!(response_id, "");
-                assert!(token_usage.is_none());
-                assert!(*can_append);
-            }
-            other => panic!("unexpected event: {other:?}"),
-        }
-    }
-
     #[tokio::test]
     async fn emits_completed_without_stream_end() {
         let completed = json!({
@@ -770,11 +671,9 @@ mod tests {
             Ok(ResponseEvent::Completed {
                 response_id,
                 token_usage,
-                can_append,
             }) => {
                 assert_eq!(response_id, "resp1");
                 assert!(token_usage.is_none());
-                assert!(!can_append);
             }
             other => panic!("unexpected event: {other:?}"),
         }
@@ -996,8 +895,7 @@ mod tests {
             &events[1],
             ResponseEvent::Completed {
                 response_id,
-                token_usage: None,
-                can_append: false
+                token_usage: None
             } if response_id == "resp-1"
         );
     }
@@ -1033,8 +931,7 @@ mod tests {
             &events[2],
             ResponseEvent::Completed {
                 response_id,
-                token_usage: None,
-                can_append: false
+                token_usage: None
             } if response_id == "resp-1"
         );
     }

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

@@ -159,11 +159,9 @@ async fn responses_stream_parses_items_and_completed_end_to_end() -> Result<()>
         ResponseEvent::Completed {
             response_id,
             token_usage,
-            can_append,
         } => {
             assert_eq!(response_id, "resp1");
             assert!(token_usage.is_none());
-            assert!(!can_append);
         }
         other => panic!("unexpected third event: {other:?}"),
     }

codex-rs/core/src/client.rs

@@ -20,9 +20,8 @@
 //!
 //! ## Retry-Budget Tradeoff
 //!
-//! V2 request prewarm is treated as the first websocket connection attempt for a turn. If it
-//! fails, normal stream retry/fallback logic handles recovery on the same turn. V1 prewarm
-//! remains connection-only.
+//! WebSocket prewarm is treated as the first websocket connection attempt for a turn. If it
+//! fails, normal stream retry/fallback logic handles recovery on the same turn.
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -43,7 +42,6 @@ use codex_api::MemorySummarizeOutput as ApiMemorySummarizeOutput;
 use codex_api::RawMemory as ApiRawMemory;
 use codex_api::RequestTelemetry;
 use codex_api::ReqwestTransport;
-use codex_api::ResponseAppendWsRequest;
 use codex_api::ResponseCreateWsRequest;
 use codex_api::ResponsesApiRequest;
 use codex_api::ResponsesClient as ApiResponsesClient;
@@ -101,32 +99,19 @@ use crate::model_provider_info::WireApi;
 use crate::tools::spec::create_tools_json_for_responses_api;
 
 pub const OPENAI_BETA_HEADER: &str = "OpenAI-Beta";
-pub const OPENAI_BETA_RESPONSES_WEBSOCKETS: &str = "responses_websockets=2026-02-04";
 pub const X_CODEX_TURN_STATE_HEADER: &str = "x-codex-turn-state";
 pub const X_CODEX_TURN_METADATA_HEADER: &str = "x-codex-turn-metadata";
 pub const X_RESPONSESAPI_INCLUDE_TIMING_METRICS_HEADER: &str =
     "x-responsesapi-include-timing-metrics";
 const RESPONSES_WEBSOCKETS_V2_BETA_HEADER_VALUE: &str = "responses_websockets=2026-02-06";
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum ResponsesWebsocketVersion {
-    V1,
-    V2,
-}
-
-pub fn ws_version_from_features(config: &Config) -> Option<ResponsesWebsocketVersion> {
-    match (
-        config
+pub fn ws_version_from_features(config: &Config) -> bool {
+    config
+        .features
+        .enabled(crate::features::Feature::ResponsesWebsockets)
+        || config
             .features
-            .enabled(crate::features::Feature::ResponsesWebsockets),
-        config
-            .features
-            .enabled(crate::features::Feature::ResponsesWebsocketsV2),
-    ) {
-        (_, true) => Some(ResponsesWebsocketVersion::V2),
-        (true, false) => Some(ResponsesWebsocketVersion::V1),
-        (false, false) => None,
-    }
+            .enabled(crate::features::Feature::ResponsesWebsocketsV2)
 }
 
 /// Session-scoped state shared by all [`ModelClient`] clones.
@@ -140,7 +125,7 @@ struct ModelClientState {
     provider: ModelProviderInfo,
     session_source: SessionSource,
     model_verbosity: Option<VerbosityConfig>,
-    responses_websocket_version: Option<ResponsesWebsocketVersion>,
+    responses_websockets_enabled_by_feature: bool,
     enable_request_compression: bool,
     include_timing_metrics: bool,
     beta_features_header: Option<String>,
@@ -180,8 +165,8 @@ pub struct ModelClient {
 /// The session establishes a Responses WebSocket connection lazily and reuses it across multiple
 /// requests within the turn. It also caches per-turn state:
 ///
-/// - The last full request, so subsequent calls can use `response.append` only when the current
-///   request is an incremental extension of the previous one.
+/// - The last full request, so subsequent calls can reuse incremental websocket request payloads
+///   only when the current request is an incremental extension of the previous one.
 /// - The `x-codex-turn-state` sticky-routing token, which must be replayed for all requests within
 ///   the same turn.
 ///
@@ -208,7 +193,6 @@ pub struct ModelClientSession {
 struct LastResponse {
     response_id: String,
     items_added: Vec<ResponseItem>,
-    can_append: bool,
 }
 
 #[derive(Debug, Default)]
@@ -235,7 +219,7 @@ impl ModelClient {
         provider: ModelProviderInfo,
         session_source: SessionSource,
         model_verbosity: Option<VerbosityConfig>,
-        responses_websocket_version: Option<ResponsesWebsocketVersion>,
+        responses_websockets_enabled_by_feature: bool,
         enable_request_compression: bool,
         include_timing_metrics: bool,
         beta_features_header: Option<String>,
@@ -247,7 +231,7 @@ impl ModelClient {
                 provider,
                 session_source,
                 model_verbosity,
-                responses_websocket_version,
+                responses_websockets_enabled_by_feature,
                 enable_request_compression,
                 include_timing_metrics,
                 beta_features_header,
@@ -391,25 +375,21 @@ impl ModelClient {
         request_telemetry
     }
 
-    /// Returns the active Responses-over-WebSocket version for this session.
+    /// Returns whether the Responses-over-WebSocket transport is active for this session.
     ///
     /// This combines provider capability and feature gating; both must be true for websocket paths
     /// to be eligible.
     ///
     /// If websockets are only enabled via model preference (no explicit feature flag), prefer the
     /// current v2 behavior.
-    pub fn active_ws_version(&self, model_info: &ModelInfo) -> Option<ResponsesWebsocketVersion> {
+    pub fn responses_websocket_enabled(&self, model_info: &ModelInfo) -> bool {
         if !self.state.provider.supports_websockets
             || self.state.disable_websockets.load(Ordering::Relaxed)
         {
-            return None;
+            return false;
         }
 
-        match self.state.responses_websocket_version {
-            Some(version) => Some(version),
-            None if model_info.prefer_websockets => Some(ResponsesWebsocketVersion::V2),
-            None => None,
-        }
+        self.state.responses_websockets_enabled_by_feature || model_info.prefer_websockets
     }
 
     /// Returns auth + provider configuration resolved from the current session auth state.
@@ -442,12 +422,10 @@ impl ModelClient {
         otel_manager: &OtelManager,
         api_provider: codex_api::Provider,
         api_auth: CoreAuthProvider,
-        ws_version: ResponsesWebsocketVersion,
         turn_state: Option<Arc<OnceLock<String>>>,
         turn_metadata_header: Option<&str>,
     ) -> std::result::Result<ApiWebSocketConnection, ApiError> {
-        let headers =
-            self.build_websocket_headers(ws_version, turn_state.as_ref(), turn_metadata_header);
+        let headers = self.build_websocket_headers(turn_state.as_ref(), turn_metadata_header);
         let websocket_telemetry = ModelClientSession::build_websocket_telemetry(otel_manager);
         ApiWebSocketResponsesClient::new(api_provider, api_auth)
             .connect(
@@ -465,7 +443,6 @@ impl ModelClient {
     /// replayed on reconnect within the same turn.
     fn build_websocket_headers(
         &self,
-        ws_version: ResponsesWebsocketVersion,
         turn_state: Option<&Arc<OnceLock<String>>>,
         turn_metadata_header: Option<&str>,
     ) -> ApiHeaderMap {
@@ -478,13 +455,9 @@ impl ModelClient {
         headers.extend(build_conversation_headers(Some(
             self.state.conversation_id.to_string(),
         )));
-        let responses_websockets_beta_header = match ws_version {
-            ResponsesWebsocketVersion::V2 => RESPONSES_WEBSOCKETS_V2_BETA_HEADER_VALUE,
-            ResponsesWebsocketVersion::V1 => OPENAI_BETA_RESPONSES_WEBSOCKETS,
-        };
         headers.insert(
             OPENAI_BETA_HEADER,
-            HeaderValue::from_static(responses_websockets_beta_header),
+            HeaderValue::from_static(RESPONSES_WEBSOCKETS_V2_BETA_HEADER_VALUE),
         );
         if self.state.include_timing_metrics {
             headers.insert(
@@ -613,8 +586,9 @@ impl ModelClientSession {
         last_response: Option<&LastResponse>,
         allow_empty_delta: bool,
     ) -> Option<Vec<ResponseItem>> {
-        // Checks whether the current request is an incremental append to the previous request.
-        // We only append when non-input request fields are unchanged and `input` is a strict
+        // Checks whether the current request is an incremental extension of the previous request.
+        // We only reuse an incremental input delta when non-input request fields are unchanged and
+        // `input` is a strict
         // extension of the previous known input. Server-returned output items are treated as part
         // of the baseline so we do not resend them.
         let previous_request = self.websocket_session.last_request.as_ref()?;
@@ -659,42 +633,26 @@ impl ModelClientSession {
         &mut self,
         payload: ResponseCreateWsRequest,
         request: &ResponsesApiRequest,
-        ws_version: ResponsesWebsocketVersion,
     ) -> ResponsesWsRequest {
         let Some(last_response) = self.get_last_response() else {
             return ResponsesWsRequest::ResponseCreate(payload);
         };
-        let allow_empty_delta = matches!(ws_version, ResponsesWebsocketVersion::V2);
-        let Some(append_items) =
-            self.get_incremental_items(request, Some(&last_response), allow_empty_delta)
+        let Some(incremental_items) =
+            self.get_incremental_items(request, Some(&last_response), true)
         else {
             return ResponsesWsRequest::ResponseCreate(payload);
         };
 
-        match ws_version {
-            ResponsesWebsocketVersion::V2 => {
-                if last_response.response_id.is_empty() {
-                    trace!("incremental request failed, no previous response id");
-                    return ResponsesWsRequest::ResponseCreate(payload);
-                }
-
-                ResponsesWsRequest::ResponseCreate(ResponseCreateWsRequest {
-                    previous_response_id: Some(last_response.response_id),
-                    input: append_items,
-                    ..payload
-                })
-            }
-            ResponsesWebsocketVersion::V1 => {
-                if !last_response.can_append {
-                    trace!("incremental request failed, can't append");
-                    return ResponsesWsRequest::ResponseCreate(payload);
-                }
-                ResponsesWsRequest::ResponseAppend(ResponseAppendWsRequest {
-                    input: append_items,
-                    client_metadata: payload.client_metadata,
-                })
-            }
+        if last_response.response_id.is_empty() {
+            trace!("incremental request failed, no previous response id");
+            return ResponsesWsRequest::ResponseCreate(payload);
         }
+
+        ResponsesWsRequest::ResponseCreate(ResponseCreateWsRequest {
+            previous_response_id: Some(last_response.response_id),
+            input: incremental_items,
+            ..payload
+        })
     }
 
     /// Opportunistically preconnects a websocket for this turn-scoped client session.
@@ -705,9 +663,9 @@ impl ModelClientSession {
         otel_manager: &OtelManager,
         model_info: &ModelInfo,
     ) -> std::result::Result<(), ApiError> {
-        let Some(ws_version) = self.client.active_ws_version(model_info) else {
+        if !self.client.responses_websocket_enabled(model_info) {
             return Ok(());
-        };
+        }
         if self.websocket_session.connection.is_some() {
             return Ok(());
         }
@@ -724,7 +682,6 @@ impl ModelClientSession {
                 otel_manager,
                 client_setup.api_provider,
                 client_setup.api_auth,
-                ws_version,
                 Some(Arc::clone(&self.turn_state)),
                 None,
             )
@@ -738,7 +695,6 @@ impl ModelClientSession {
         otel_manager: &OtelManager,
         api_provider: codex_api::Provider,
         api_auth: CoreAuthProvider,
-        ws_version: ResponsesWebsocketVersion,
         turn_metadata_header: Option<&str>,
         options: &ApiResponsesOptions,
     ) -> std::result::Result<&ApiWebSocketConnection, ApiError> {
@@ -760,7 +716,6 @@ impl ModelClientSession {
                     otel_manager,
                     api_provider,
                     api_auth,
-                    ws_version,
                     Some(turn_state),
                     turn_metadata_header,
                 )
@@ -862,7 +817,6 @@ impl ModelClientSession {
         &mut self,
         prompt: &Prompt,
         model_info: &ModelInfo,
-        ws_version: ResponsesWebsocketVersion,
         otel_manager: &OtelManager,
         effort: Option<ReasoningEffortConfig>,
         summary: ReasoningSummaryConfig,
@@ -901,7 +855,6 @@ impl ModelClientSession {
                     otel_manager,
                     client_setup.api_provider,
                     client_setup.api_auth,
-                    ws_version,
                     turn_metadata_header,
                     &options,
                 )
@@ -922,7 +875,7 @@ impl ModelClientSession {
                 Err(err) => return Err(map_api_error(err)),
             }
 
-            let ws_request = self.prepare_websocket_request(ws_payload, &request, ws_version);
+            let ws_request = self.prepare_websocket_request(ws_payload, &request);
             self.websocket_session.last_request = Some(request);
             let stream_result = self
                 .websocket_session
@@ -971,17 +924,10 @@ impl ModelClientSession {
         service_tier: Option<ServiceTier>,
         turn_metadata_header: Option<&str>,
     ) -> Result<()> {
-        let Some(ws_version) = self.client.active_ws_version(model_info) else {
-            return Ok(());
-        };
-        if self.websocket_session.last_request.is_some() {
+        if !self.client.responses_websocket_enabled(model_info) {
             return Ok(());
         }
-
-        if matches!(ws_version, ResponsesWebsocketVersion::V1) {
-            self.preconnect_websocket(otel_manager, model_info)
-                .await
-                .map_err(map_api_error)?;
+        if self.websocket_session.last_request.is_some() {
             return Ok(());
         }
 
@@ -989,7 +935,6 @@ impl ModelClientSession {
             .stream_responses_websocket(
                 prompt,
                 model_info,
-                ws_version,
                 otel_manager,
                 effort,
                 summary,
@@ -1038,12 +983,11 @@ impl ModelClientSession {
         let wire_api = self.client.state.provider.wire_api;
         match wire_api {
             WireApi::Responses => {
-                if let Some(ws_version) = self.client.active_ws_version(model_info) {
+                if self.client.responses_websocket_enabled(model_info) {
                     match self
                         .stream_responses_websocket(
                             prompt,
                             model_info,
-                            ws_version,
                             otel_manager,
                             effort,
                             summary,
@@ -1085,7 +1029,7 @@ impl ModelClientSession {
         otel_manager: &OtelManager,
         model_info: &ModelInfo,
     ) -> bool {
-        let websocket_enabled = self.client.active_ws_version(model_info).is_some();
+        let websocket_enabled = self.client.responses_websocket_enabled(model_info);
         let activated = self.activate_http_fallback(websocket_enabled);
         if activated {
             warn!("falling back to HTTP");
@@ -1183,7 +1127,6 @@ where
                 Ok(ResponseEvent::Completed {
                     response_id,
                     token_usage,
-                    can_append,
                 }) => {
                     if let Some(usage) = &token_usage {
                         otel_manager.sse_event_completed(
@@ -1198,14 +1141,12 @@ where
                         let _ = sender.send(LastResponse {
                             response_id: response_id.clone(),
                             items_added: std::mem::take(&mut items_added),
-                            can_append,
                         });
                     }
                     if tx_event
                         .send(Ok(ResponseEvent::Completed {
                             response_id,
                             token_usage,
-                            can_append,
                         }))
                         .await
                         .is_err()
@@ -1335,7 +1276,7 @@ mod tests {
             provider,
             session_source,
             None,
-            None,
+            false,
             false,
             false,
             None,

codex-rs/core/src/codex.rs

@@ -5572,12 +5572,10 @@ async fn run_sampling_request(
             // transient reconnect messages. In debug builds, keep full visibility for diagnosis.
             let report_error = retries > 1
                 || cfg!(debug_assertions)
-                || sess
+                || !sess
                     .services
                     .model_client
-                    .active_ws_version(&turn_context.model_info)
-                    .is_none();
-
+                    .responses_websocket_enabled(&turn_context.model_info);
             if report_error {
                 // Surface retry information to any UI/front‑end so the
                 // user understands what is happening instead of staring
@@ -6418,7 +6416,6 @@ async fn try_run_sampling_request(
             ResponseEvent::Completed {
                 response_id: _,
                 token_usage,
-                can_append: _,
             } => {
                 flush_assistant_text_segments_all(
                     &sess,
@@ -9194,7 +9191,7 @@ mod tests {
         handlers::run_user_shell_command(&session, "sub-id".to_string(), "echo shell".to_string())
             .await;
 
-        let deadline = StdDuration::from_secs(5);
+        let deadline = StdDuration::from_secs(15);
         let start = std::time::Instant::now();
         loop {
             let remaining = deadline.saturating_sub(start.elapsed());

codex-rs/core/src/compact.rs

@@ -109,7 +109,8 @@ async fn run_compact_task_inner(
     let max_retries = turn_context.provider.stream_max_retries();
     let mut retries = 0;
     let mut client_session = sess.services.model_client.new_session();
-    // Reuse one client session so turn-scoped state (sticky routing, websocket append tracking)
+    // Reuse one client session so turn-scoped state (sticky routing, websocket incremental
+    // request tracking)
     // survives retries within this compact turn.
 
     loop {

codex-rs/core/src/lib.rs

@@ -147,7 +147,6 @@ pub(crate) use codex_shell_command::powershell;
 
 pub use client::ModelClient;
 pub use client::ModelClientSession;
-pub use client::ResponsesWebsocketVersion;
 pub use client::X_CODEX_TURN_METADATA_HEADER;
 pub use client::ws_version_from_features;
 pub use client_common::Prompt;

codex-rs/core/src/tools/runtimes/shell.rs

@@ -5,7 +5,7 @@ Executes shell requests under the orchestrator: asks for approval when needed,
 builds a CommandSpec, and runs it under the current SandboxAttempt.
 */
 #[cfg(unix)]
-mod unix_escalation;
+pub(crate) mod unix_escalation;
 
 use crate::command_canonicalization::canonicalize_command_for_approval;
 use crate::exec::ExecToolCallOutput;

codex-rs/core/src/tools/runtimes/shell/unix_escalation.rs

@@ -6,6 +6,7 @@ use crate::exec::ExecToolCallOutput;
 use crate::exec::SandboxType;
 use crate::exec::is_likely_sandbox_denied;
 use crate::features::Feature;
+use crate::sandboxing::ExecRequest;
 use crate::sandboxing::SandboxPermissions;
 use crate::sandboxing::sandbox_policy_with_additional_permissions;
 use crate::shell::ShellType;
@@ -37,6 +38,7 @@ use codex_shell_escalation::EscalationDecision;
 use codex_shell_escalation::EscalationExecution;
 use codex_shell_escalation::EscalationPermissions;
 use codex_shell_escalation::EscalationPolicy;
+use codex_shell_escalation::EscalationSession;
 use codex_shell_escalation::ExecParams;
 use codex_shell_escalation::ExecResult;
 use codex_shell_escalation::Permissions as EscalatedPermissions;
@@ -52,6 +54,11 @@ use tokio::sync::RwLock;
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
+pub(crate) struct PreparedUnifiedExecZshFork {
+    pub(crate) exec_request: ExecRequest,
+    pub(crate) escalation_session: EscalationSession,
+}
+
 pub(super) async fn try_run_zsh_fork(
     req: &ShellRequest,
     attempt: &SandboxAttempt<'_>,
@@ -96,7 +103,7 @@ pub(super) async fn try_run_zsh_fork(
         justification,
         arg0,
     } = sandbox_exec_request;
-    let ParsedShellCommand { script, login } = extract_shell_script(&command)?;
+    let ParsedShellCommand { script, login, .. } = extract_shell_script(&command)?;
     let effective_timeout = Duration::from_millis(
         req.timeout_ms
             .unwrap_or(crate::exec::DEFAULT_EXEC_COMMAND_TIMEOUT_MS),
@@ -173,6 +180,103 @@ pub(super) async fn try_run_zsh_fork(
     map_exec_result(attempt.sandbox, exec_result).map(Some)
 }
 
+pub(crate) async fn prepare_unified_exec_zsh_fork(
+    req: &crate::tools::runtimes::unified_exec::UnifiedExecRequest,
+    attempt: &SandboxAttempt<'_>,
+    ctx: &ToolCtx,
+    exec_request: ExecRequest,
+) -> Result<Option<PreparedUnifiedExecZshFork>, ToolError> {
+    let Some(shell_zsh_path) = ctx.session.services.shell_zsh_path.as_ref() else {
+        tracing::warn!("ZshFork backend specified, but shell_zsh_path is not configured.");
+        return Ok(None);
+    };
+    if !ctx.session.features().enabled(Feature::ShellZshFork) {
+        tracing::warn!("ZshFork backend specified, but ShellZshFork feature is not enabled.");
+        return Ok(None);
+    }
+    if !matches!(ctx.session.user_shell().shell_type, ShellType::Zsh) {
+        tracing::warn!("ZshFork backend specified, but user shell is not Zsh.");
+        return Ok(None);
+    }
+
+    let parsed = match extract_shell_script(&exec_request.command) {
+        Ok(parsed) => parsed,
+        Err(err) => {
+            tracing::warn!("ZshFork unified exec fallback: {err:?}");
+            return Ok(None);
+        }
+    };
+    if parsed.program != shell_zsh_path.to_string_lossy() {
+        tracing::warn!(
+            "ZshFork backend specified, but unified exec command targets `{}` instead of `{}`.",
+            parsed.program,
+            shell_zsh_path.display(),
+        );
+        return Ok(None);
+    }
+
+    let exec_policy = Arc::new(RwLock::new(
+        ctx.session.services.exec_policy.current().as_ref().clone(),
+    ));
+    let command_executor = CoreShellCommandExecutor {
+        command: exec_request.command.clone(),
+        cwd: exec_request.cwd.clone(),
+        sandbox_policy: exec_request.sandbox_policy.clone(),
+        sandbox: exec_request.sandbox,
+        env: exec_request.env.clone(),
+        network: exec_request.network.clone(),
+        windows_sandbox_level: exec_request.windows_sandbox_level,
+        sandbox_permissions: exec_request.sandbox_permissions,
+        justification: exec_request.justification.clone(),
+        arg0: exec_request.arg0.clone(),
+        sandbox_policy_cwd: ctx.turn.cwd.clone(),
+        macos_seatbelt_profile_extensions: ctx
+            .turn
+            .config
+            .permissions
+            .macos_seatbelt_profile_extensions
+            .clone(),
+        codex_linux_sandbox_exe: ctx.turn.codex_linux_sandbox_exe.clone(),
+        use_linux_sandbox_bwrap: ctx.turn.features.enabled(Feature::UseLinuxSandboxBwrap),
+    };
+    let main_execve_wrapper_exe = ctx
+        .session
+        .services
+        .main_execve_wrapper_exe
+        .clone()
+        .ok_or_else(|| {
+            ToolError::Rejected(
+                "zsh fork feature enabled, but execve wrapper is not configured".to_string(),
+            )
+        })?;
+    let escalation_policy = CoreShellActionProvider {
+        policy: Arc::clone(&exec_policy),
+        session: Arc::clone(&ctx.session),
+        turn: Arc::clone(&ctx.turn),
+        call_id: ctx.call_id.clone(),
+        approval_policy: ctx.turn.approval_policy.value(),
+        sandbox_policy: attempt.policy.clone(),
+        sandbox_permissions: req.sandbox_permissions,
+        prompt_permissions: req.additional_permissions.clone(),
+        stopwatch: Stopwatch::unlimited(),
+    };
+
+    let escalate_server = EscalateServer::new(
+        shell_zsh_path.clone(),
+        main_execve_wrapper_exe,
+        escalation_policy,
+    );
+    let escalation_session = escalate_server
+        .start_session(Arc::new(command_executor))
+        .map_err(|err| ToolError::Rejected(err.to_string()))?;
+    let mut exec_request = exec_request;
+    exec_request.env.extend(escalation_session.env().clone());
+    Ok(Some(PreparedUnifiedExecZshFork {
+        exec_request,
+        escalation_session,
+    }))
+}
+
 struct CoreShellActionProvider {
     policy: Arc<RwLock<Policy>>,
     session: Arc<crate::codex::Session>,
@@ -858,6 +962,7 @@ impl CoreShellCommandExecutor {
 
 #[derive(Debug, Eq, PartialEq)]
 struct ParsedShellCommand {
+    program: String,
     script: String,
     login: bool,
 }
@@ -866,12 +971,20 @@ fn extract_shell_script(command: &[String]) -> Result<ParsedShellCommand, ToolEr
     // Commands reaching zsh-fork can be wrapped by environment/sandbox helpers, so
     // we search for the first `-c`/`-lc` triple anywhere in the argv rather
     // than assuming it is the first positional form.
-    if let Some((script, login)) = command.windows(3).find_map(|parts| match parts {
-        [_, flag, script] if flag == "-c" => Some((script.to_owned(), false)),
-        [_, flag, script] if flag == "-lc" => Some((script.to_owned(), true)),
+    if let Some((program, script, login)) = command.windows(3).find_map(|parts| match parts {
+        [program, flag, script] if flag == "-c" => {
+            Some((program.to_owned(), script.to_owned(), false))
+        }
+        [program, flag, script] if flag == "-lc" => {
+            Some((program.to_owned(), script.to_owned(), true))
+        }
         _ => None,
     }) {
-        return Ok(ParsedShellCommand { script, login });
+        return Ok(ParsedShellCommand {
+            program,
+            script,
+            login,
+        });
     }
 
     Err(ToolError::Rejected(

codex-rs/core/src/tools/runtimes/shell/unix_escalation_tests.rs

@@ -64,6 +64,7 @@ fn extract_shell_script_preserves_login_flag() {
     assert_eq!(
         extract_shell_script(&["/bin/zsh".into(), "-lc".into(), "echo hi".into()]).unwrap(),
         ParsedShellCommand {
+            program: "/bin/zsh".to_string(),
             script: "echo hi".to_string(),
             login: true,
         }
@@ -71,6 +72,7 @@ fn extract_shell_script_preserves_login_flag() {
     assert_eq!(
         extract_shell_script(&["/bin/zsh".into(), "-c".into(), "echo hi".into()]).unwrap(),
         ParsedShellCommand {
+            program: "/bin/zsh".to_string(),
             script: "echo hi".to_string(),
             login: false,
         }
@@ -89,6 +91,7 @@ fn extract_shell_script_supports_wrapped_command_prefixes() {
         ])
         .unwrap(),
         ParsedShellCommand {
+            program: "/bin/zsh".to_string(),
             script: "echo hello".to_string(),
             login: true,
         }
@@ -105,6 +108,7 @@ fn extract_shell_script_supports_wrapped_command_prefixes() {
         ])
         .unwrap(),
         ParsedShellCommand {
+            program: "/bin/zsh".to_string(),
             script: "pwd".to_string(),
             login: false,
         }

codex-rs/core/src/tools/runtimes/unified_exec.rs

@@ -16,6 +16,8 @@ use crate::tools::network_approval::NetworkApprovalMode;
 use crate::tools::network_approval::NetworkApprovalSpec;
 use crate::tools::runtimes::build_command_spec;
 use crate::tools::runtimes::maybe_wrap_shell_lc_with_snapshot;
+#[cfg(unix)]
+use crate::tools::runtimes::shell::unix_escalation;
 use crate::tools::sandboxing::Approvable;
 use crate::tools::sandboxing::ApprovalCtx;
 use crate::tools::sandboxing::ExecApprovalRequirement;
@@ -28,6 +30,7 @@ use crate::tools::sandboxing::ToolError;
 use crate::tools::sandboxing::ToolRuntime;
 use crate::tools::sandboxing::sandbox_override_for_first_attempt;
 use crate::tools::sandboxing::with_cached_approval;
+use crate::tools::spec::UnifiedExecBackendConfig;
 use crate::unified_exec::UnifiedExecError;
 use crate::unified_exec::UnifiedExecProcess;
 use crate::unified_exec::UnifiedExecProcessManager;
@@ -63,11 +66,12 @@ pub struct UnifiedExecApprovalKey {
 
 pub struct UnifiedExecRuntime<'a> {
     manager: &'a UnifiedExecProcessManager,
+    backend: UnifiedExecBackendConfig,
 }
 
 impl<'a> UnifiedExecRuntime<'a> {
-    pub fn new(manager: &'a UnifiedExecProcessManager) -> Self {
-        Self { manager }
+    pub fn new(manager: &'a UnifiedExecProcessManager, backend: UnifiedExecBackendConfig) -> Self {
+        Self { manager, backend }
     }
 }
 
@@ -194,11 +198,59 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecProcess> for UnifiedExecRunt
             req.justification.clone(),
         )
         .map_err(|_| ToolError::Rejected("missing command line for PTY".to_string()))?;
+        #[cfg(unix)]
+        if self.backend == UnifiedExecBackendConfig::ZshFork {
+            let exec_env = attempt
+                .env_for(spec, req.network.as_ref())
+                .map_err(|err| ToolError::Codex(err.into()))?;
+            match unix_escalation::prepare_unified_exec_zsh_fork(req, attempt, ctx, exec_env)
+                .await?
+            {
+                Some(prepared) => {
+                    let unix_escalation::PreparedUnifiedExecZshFork {
+                        exec_request,
+                        escalation_session,
+                    } = prepared;
+                    return self
+                        .manager
+                        .open_session_with_exec_env(
+                            &exec_request,
+                            req.tty,
+                            Some(escalation_session),
+                        )
+                        .await
+                        .map_err(|err| match err {
+                            UnifiedExecError::SandboxDenied { output, .. } => {
+                                ToolError::Codex(CodexErr::Sandbox(SandboxErr::Denied {
+                                    output: Box::new(output),
+                                    network_policy_decision: None,
+                                }))
+                            }
+                            other => ToolError::Rejected(other.to_string()),
+                        });
+                }
+                None => {
+                    tracing::warn!(
+                        "UnifiedExec ZshFork backend specified, but conditions for using it were not met, falling back to direct execution",
+                    );
+                }
+            }
+        }
+        let spec = build_command_spec(
+            &command,
+            &req.cwd,
+            &env,
+            ExecExpiration::DefaultTimeout,
+            req.sandbox_permissions,
+            req.additional_permissions.clone(),
+            req.justification.clone(),
+        )
+        .map_err(|_| ToolError::Rejected("missing command line for PTY".to_string()))?;
         let exec_env = attempt
             .env_for(spec, req.network.as_ref())
             .map_err(|err| ToolError::Codex(err.into()))?;
         self.manager
-            .open_session_with_exec_env(&exec_env, req.tty)
+            .open_session_with_exec_env(&exec_env, req.tty, None)
             .await
             .map_err(|err| match err {
                 UnifiedExecError::SandboxDenied { output, .. } => {

codex-rs/core/src/tools/spec.rs

@@ -42,10 +42,17 @@ pub enum ShellCommandBackendConfig {
     ZshFork,
 }
 
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum UnifiedExecBackendConfig {
+    Direct,
+    ZshFork,
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct ToolsConfig {
     pub shell_type: ConfigShellToolType,
     shell_command_backend: ShellCommandBackendConfig,
+    pub unified_exec_backend: UnifiedExecBackendConfig,
     pub allow_login_shell: bool,
     pub apply_patch_tool_type: Option<ApplyPatchToolType>,
     pub web_search_mode: Option<WebSearchMode>,
@@ -94,6 +101,12 @@ impl ToolsConfig {
             } else {
                 ShellCommandBackendConfig::Classic
             };
+        let unified_exec_backend =
+            if features.enabled(Feature::ShellTool) && features.enabled(Feature::ShellZshFork) {
+                UnifiedExecBackendConfig::ZshFork
+            } else {
+                UnifiedExecBackendConfig::Direct
+            };
 
         let shell_type = if !features.enabled(Feature::ShellTool) {
             ConfigShellToolType::Disabled
@@ -132,6 +145,7 @@ impl ToolsConfig {
         Self {
             shell_type,
             shell_command_backend,
+            unified_exec_backend,
             allow_login_shell: true,
             apply_patch_tool_type,
             web_search_mode: *web_search_mode,
@@ -2727,6 +2741,10 @@ mod tests {
             tools_config.shell_command_backend,
             ShellCommandBackendConfig::ZshFork
         );
+        assert_eq!(
+            tools_config.unified_exec_backend,
+            UnifiedExecBackendConfig::ZshFork
+        );
     }
 
     #[test]

codex-rs/core/src/unified_exec/process.rs

@@ -17,6 +17,7 @@ use crate::exec::StreamOutput;
 use crate::exec::is_likely_sandbox_denied;
 use crate::truncate::TruncationPolicy;
 use crate::truncate::formatted_truncate_text;
+use codex_shell_escalation::EscalationSession;
 use codex_utils_pty::ExecCommandSession;
 use codex_utils_pty::SpawnedPty;
 
@@ -44,6 +45,7 @@ pub(crate) struct UnifiedExecProcess {
     output_drained: Arc<Notify>,
     output_task: JoinHandle<()>,
     sandbox_type: SandboxType,
+    _escalation_session: Option<EscalationSession>,
 }
 
 impl UnifiedExecProcess {
@@ -51,6 +53,7 @@ impl UnifiedExecProcess {
         process_handle: ExecCommandSession,
         initial_output_rx: tokio::sync::broadcast::Receiver<Vec<u8>>,
         sandbox_type: SandboxType,
+        escalation_session: Option<EscalationSession>,
     ) -> Self {
         let output_buffer = Arc::new(Mutex::new(HeadTailBuffer::default()));
         let output_notify = Arc::new(Notify::new());
@@ -92,6 +95,7 @@ impl UnifiedExecProcess {
             output_drained,
             output_task,
             sandbox_type,
+            _escalation_session: escalation_session,
         }
     }
 
@@ -196,13 +200,14 @@ impl UnifiedExecProcess {
     pub(super) async fn from_spawned(
         spawned: SpawnedPty,
         sandbox_type: SandboxType,
+        escalation_session: Option<EscalationSession>,
     ) -> Result<Self, UnifiedExecError> {
         let SpawnedPty {
             session: process_handle,
             output_rx,
             mut exit_rx,
         } = spawned;
-        let managed = Self::new(process_handle, output_rx, sandbox_type);
+        let managed = Self::new(process_handle, output_rx, sandbox_type, escalation_session);
 
         let exit_ready = matches!(exit_rx.try_recv(), Ok(_) | Err(TryRecvError::Closed));
 

codex-rs/core/src/unified_exec/process_manager.rs

@@ -51,6 +51,7 @@ use crate::unified_exec::process::OutputBuffer;
 use crate::unified_exec::process::OutputHandles;
 use crate::unified_exec::process::UnifiedExecProcess;
 use crate::unified_exec::resolve_max_tokens;
+use codex_shell_escalation::EscalationSession;
 
 const UNIFIED_EXEC_ENV: [(&str, &str); 10] = [
     ("NO_COLOR", "1"),
@@ -528,6 +529,7 @@ impl UnifiedExecProcessManager {
         &self,
         env: &ExecRequest,
         tty: bool,
+        escalation_session: Option<EscalationSession>,
     ) -> Result<UnifiedExecProcess, UnifiedExecError> {
         let (program, args) = env
             .command
@@ -555,7 +557,7 @@ impl UnifiedExecProcessManager {
         };
         let spawned =
             spawn_result.map_err(|err| UnifiedExecError::create_process(err.to_string()))?;
-        UnifiedExecProcess::from_spawned(spawned, env.sandbox).await
+        UnifiedExecProcess::from_spawned(spawned, env.sandbox, escalation_session).await
     }
 
     pub(super) async fn open_session_with_sandbox(
@@ -569,7 +571,8 @@ impl UnifiedExecProcessManager {
             Some(context.session.conversation_id),
         ));
         let mut orchestrator = ToolOrchestrator::new();
-        let mut runtime = UnifiedExecRuntime::new(self);
+        let mut runtime =
+            UnifiedExecRuntime::new(self, context.turn.tools_config.unified_exec_backend);
         let exec_approval_requirement = context
             .session
             .services

codex-rs/core/tests/common/responses.rs

@@ -582,25 +582,6 @@ pub fn ev_completed(id: &str) -> Value {
     })
 }
 
-pub fn ev_done() -> Value {
-    serde_json::json!({
-        "type": "response.done",
-        "response": {
-            "usage": {"input_tokens":0,"input_tokens_details":null,"output_tokens":0,"output_tokens_details":null,"total_tokens":0}
-        }
-    })
-}
-
-pub fn ev_done_with_id(id: &str) -> Value {
-    serde_json::json!({
-        "type": "response.done",
-        "response": {
-            "id": id,
-            "usage": {"input_tokens":0,"input_tokens_details":null,"output_tokens":0,"output_tokens_details":null,"total_tokens":0}
-        }
-    })
-}
-
 /// Convenience: SSE event for a created response with a specific id.
 pub fn ev_response_created(id: &str) -> Value {
     serde_json::json!({

codex-rs/core/tests/responses_headers.rs

@@ -6,7 +6,6 @@ use codex_core::ModelClient;
 use codex_core::ModelProviderInfo;
 use codex_core::Prompt;
 use codex_core::ResponseEvent;
-use codex_core::ResponsesWebsocketVersion;
 use codex_core::WireApi;
 use codex_otel::OtelManager;
 use codex_otel::TelemetryAuthMode;
@@ -92,7 +91,7 @@ async fn responses_stream_includes_subagent_header_on_review() {
         provider.clone(),
         session_source,
         config.model_verbosity,
-        None::<ResponsesWebsocketVersion>,
+        false,
         false,
         false,
         None,
@@ -205,7 +204,7 @@ async fn responses_stream_includes_subagent_header_on_other() {
         provider.clone(),
         session_source,
         config.model_verbosity,
-        None::<ResponsesWebsocketVersion>,
+        false,
         false,
         false,
         None,
@@ -317,7 +316,7 @@ async fn responses_respects_model_info_overrides_from_config() {
         provider.clone(),
         session_source,
         config.model_verbosity,
-        None::<ResponsesWebsocketVersion>,
+        false,
         false,
         false,
         None,

codex-rs/core/tests/suite/agent_websocket.rs

@@ -4,8 +4,6 @@ use codex_protocol::config_types::ServiceTier;
 use core_test_support::responses::WebSocketConnectionConfig;
 use core_test_support::responses::ev_assistant_message;
 use core_test_support::responses::ev_completed;
-use core_test_support::responses::ev_done;
-use core_test_support::responses::ev_done_with_id;
 use core_test_support::responses::ev_response_created;
 use core_test_support::responses::ev_shell_command_call;
 use core_test_support::responses::start_websocket_server;
@@ -27,7 +25,7 @@ async fn websocket_test_codex_shell_chain() -> Result<()> {
         vec![
             ev_response_created("resp-1"),
             ev_shell_command_call(call_id, "echo websocket"),
-            ev_done(),
+            ev_completed("resp-1"),
         ],
         vec![
             ev_response_created("resp-2"),
@@ -59,36 +57,30 @@ async fn websocket_test_codex_shell_chain() -> Result<()> {
         .body_json();
 
     assert_eq!(first_turn["type"].as_str(), Some("response.create"));
-    assert_eq!(second_turn["type"].as_str(), Some("response.append"));
+    assert_eq!(second_turn["type"].as_str(), Some("response.create"));
 
-    let append_items = second_turn
+    let input_items = second_turn
         .get("input")
         .and_then(Value::as_array)
-        .expect("response.append input array");
-    assert!(!append_items.is_empty());
-
-    let output_item = append_items
-        .iter()
-        .find(|item| item.get("type").and_then(Value::as_str) == Some("function_call_output"))
-        .expect("function_call_output in append");
-    assert_eq!(
-        output_item.get("call_id").and_then(Value::as_str),
-        Some(call_id)
-    );
+        .expect("second response.create input array");
+    assert!(!input_items.is_empty());
 
     server.shutdown().await;
     Ok(())
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn websocket_first_turn_uses_preconnect_and_create() -> Result<()> {
+async fn websocket_first_turn_uses_startup_prewarm_and_create() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
-    let server = start_websocket_server(vec![vec![vec![
-        ev_response_created("resp-1"),
-        ev_assistant_message("msg-1", "hello"),
-        ev_completed("resp-1"),
-    ]]])
+    let server = start_websocket_server(vec![vec![
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
+        vec![
+            ev_response_created("resp-1"),
+            ev_assistant_message("msg-1", "hello"),
+            ev_completed("resp-1"),
+        ],
+    ]])
     .await;
 
     let mut builder = test_codex();
@@ -101,11 +93,14 @@ async fn websocket_first_turn_uses_preconnect_and_create() -> Result<()> {
 
     assert_eq!(server.handshakes().len(), 1);
     let connection = server.single_connection();
-    assert_eq!(connection.len(), 1);
-    let turn = connection
+    assert_eq!(connection.len(), 2);
+    let warmup = connection
         .first()
-        .expect("missing turn request")
+        .expect("missing warmup request")
         .body_json();
+    let turn = connection.get(1).expect("missing turn request").body_json();
+    assert_eq!(warmup["type"].as_str(), Some("response.create"));
+    assert_eq!(warmup["generate"].as_bool(), Some(false));
     assert!(
         turn["tools"]
             .as_array()
@@ -119,15 +114,18 @@ async fn websocket_first_turn_uses_preconnect_and_create() -> Result<()> {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn websocket_first_turn_handles_handshake_delay_with_preconnect() -> Result<()> {
+async fn websocket_first_turn_handles_handshake_delay_with_startup_prewarm() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
     let server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
-        requests: vec![vec![
-            ev_response_created("resp-1"),
-            ev_assistant_message("msg-1", "hello"),
-            ev_completed("resp-1"),
-        ]],
+        requests: vec![
+            vec![ev_response_created("warm-1"), ev_completed("warm-1")],
+            vec![
+                ev_response_created("resp-1"),
+                ev_assistant_message("msg-1", "hello"),
+                ev_completed("resp-1"),
+            ],
+        ],
         response_headers: Vec::new(),
         // Delay handshake so turn processing must tolerate websocket startup latency.
         accept_delay: Some(Duration::from_millis(150)),
@@ -144,11 +142,14 @@ async fn websocket_first_turn_handles_handshake_delay_with_preconnect() -> Resul
 
     assert_eq!(server.handshakes().len(), 1);
     let connection = server.single_connection();
-    assert_eq!(connection.len(), 1);
-    let turn = connection
+    assert_eq!(connection.len(), 2);
+    let warmup = connection
         .first()
-        .expect("missing turn request")
+        .expect("missing warmup request")
         .body_json();
+    let turn = connection.get(1).expect("missing turn request").body_json();
+    assert_eq!(warmup["type"].as_str(), Some("response.create"));
+    assert_eq!(warmup["generate"].as_bool(), Some(false));
     assert!(
         turn["tools"]
             .as_array()
@@ -167,7 +168,7 @@ async fn websocket_v2_test_codex_shell_chain() -> Result<()> {
 
     let call_id = "shell-command-call";
     let server = start_websocket_server(vec![vec![
-        vec![ev_response_created("warm-1"), ev_done_with_id("warm-1")],
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
         vec![
             ev_response_created("resp-1"),
             ev_shell_command_call(call_id, "echo websocket"),
@@ -251,7 +252,7 @@ async fn websocket_v2_first_turn_uses_updated_fast_tier_after_startup_prewarm()
     skip_if_no_network!(Ok(()));
 
     let server = start_websocket_server(vec![vec![
-        vec![ev_response_created("warm-1"), ev_done_with_id("warm-1")],
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "fast"),
@@ -300,7 +301,7 @@ async fn websocket_v2_first_turn_drops_fast_tier_after_startup_prewarm() -> Resu
     skip_if_no_network!(Ok(()));
 
     let server = start_websocket_server(vec![vec![
-        vec![ev_response_created("warm-1"), ev_done_with_id("warm-1")],
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "standard"),
@@ -349,7 +350,7 @@ async fn websocket_v2_next_turn_uses_updated_service_tier() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
     let server = start_websocket_server(vec![vec![
-        vec![ev_response_created("warm-1"), ev_done_with_id("warm-1")],
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "fast"),

codex-rs/core/tests/suite/client.rs

@@ -4,7 +4,6 @@ use codex_core::ModelProviderInfo;
 use codex_core::NewThread;
 use codex_core::Prompt;
 use codex_core::ResponseEvent;
-use codex_core::ResponsesWebsocketVersion;
 use codex_core::ThreadManager;
 use codex_core::WireApi;
 use codex_core::auth::AuthCredentialsStoreMode;
@@ -1643,7 +1642,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
         provider.clone(),
         SessionSource::Exec,
         config.model_verbosity,
-        None::<ResponsesWebsocketVersion>,
+        false,
         false,
         false,
         None,

codex-rs/core/tests/suite/client_websockets.rs

@@ -31,8 +31,6 @@ use core_test_support::responses::WebSocketConnectionConfig;
 use core_test_support::responses::WebSocketTestServer;
 use core_test_support::responses::ev_assistant_message;
 use core_test_support::responses::ev_completed;
-use core_test_support::responses::ev_done;
-use core_test_support::responses::ev_done_with_id;
 use core_test_support::responses::ev_response_created;
 use core_test_support::responses::start_websocket_server;
 use core_test_support::responses::start_websocket_server_with_headers;
@@ -50,7 +48,6 @@ use tracing_test::traced_test;
 
 const MODEL: &str = "gpt-5.2-codex";
 const OPENAI_BETA_HEADER: &str = "OpenAI-Beta";
-const OPENAI_BETA_RESPONSES_WEBSOCKETS: &str = "responses_websockets=2026-02-04";
 const WS_V2_BETA_HEADER_VALUE: &str = "responses_websockets=2026-02-06";
 
 struct WebsocketTestHarness {
@@ -89,7 +86,7 @@ async fn responses_websocket_streams_request() {
     let handshake = server.single_handshake();
     assert_eq!(
         handshake.header(OPENAI_BETA_HEADER),
-        Some(OPENAI_BETA_RESPONSES_WEBSOCKETS.to_string())
+        Some(WS_V2_BETA_HEADER_VALUE.to_string())
     );
 
     server.shutdown().await;
@@ -125,7 +122,7 @@ async fn responses_websocket_request_prewarm_reuses_connection() {
     skip_if_no_network!();
 
     let server = start_websocket_server(vec![vec![
-        vec![ev_response_created("warm-1"), ev_done_with_id("warm-1")],
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
         vec![ev_response_created("resp-1"), ev_completed("resp-1")],
     ]])
     .await;
@@ -244,7 +241,7 @@ async fn responses_websocket_request_prewarm_is_reused_even_with_header_changes(
     skip_if_no_network!();
 
     let server = start_websocket_server(vec![vec![
-        vec![ev_response_created("warm-1"), ev_done_with_id("warm-1")],
+        vec![ev_response_created("warm-1"), ev_completed("warm-1")],
         vec![ev_response_created("resp-1"), ev_completed("resp-1")],
     ]])
     .await;
@@ -344,13 +341,6 @@ async fn responses_websocket_prewarm_uses_v2_when_model_prefers_websockets_and_f
             .map(str::trim)
             .any(|value| value == WS_V2_BETA_HEADER_VALUE)
     );
-    assert!(
-        !openai_beta_header
-            .split(',')
-            .map(str::trim)
-            .any(|value| value == OPENAI_BETA_RESPONSES_WEBSOCKETS)
-    );
-
     stream_until_complete(&mut client_session, &harness, &prompt).await;
     assert_eq!(server.handshakes().len(), 1);
     let connection = server.single_connection();
@@ -404,13 +394,6 @@ async fn responses_websocket_preconnect_runs_when_only_v2_feature_enabled() {
             .map(str::trim)
             .any(|value| value == WS_V2_BETA_HEADER_VALUE)
     );
-    assert!(
-        !openai_beta_header
-            .split(',')
-            .map(str::trim)
-            .any(|value| value == OPENAI_BETA_RESPONSES_WEBSOCKETS)
-    );
-
     server.shutdown().await;
 }
 
@@ -422,7 +405,7 @@ async fn responses_websocket_v2_requests_use_v2_when_model_prefers_websockets()
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "assistant output"),
-            ev_done_with_id("resp-1"),
+            ev_completed("resp-1"),
         ],
         vec![ev_response_created("resp-2"), ev_completed("resp-2")],
     ]])
@@ -460,13 +443,6 @@ async fn responses_websocket_v2_requests_use_v2_when_model_prefers_websockets()
             .map(str::trim)
             .any(|value| value == WS_V2_BETA_HEADER_VALUE)
     );
-    assert!(
-        !openai_beta_header
-            .split(',')
-            .map(str::trim)
-            .any(|value| value == OPENAI_BETA_RESPONSES_WEBSOCKETS)
-    );
-
     server.shutdown().await;
 }
 
@@ -478,7 +454,7 @@ async fn responses_websocket_v2_incremental_requests_are_reused_across_turns() {
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "assistant output"),
-            ev_done_with_id("resp-1"),
+            ev_completed("resp-1"),
         ],
         vec![ev_response_created("resp-2"), ev_completed("resp-2")],
     ]])
@@ -522,7 +498,7 @@ async fn responses_websocket_v2_wins_when_both_features_enabled() {
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "assistant output"),
-            ev_done_with_id("resp-1"),
+            ev_completed("resp-1"),
         ],
         vec![ev_response_created("resp-2"), ev_completed("resp-2")],
     ]])
@@ -560,13 +536,6 @@ async fn responses_websocket_v2_wins_when_both_features_enabled() {
             .map(str::trim)
             .any(|value| value == WS_V2_BETA_HEADER_VALUE)
     );
-    assert!(
-        !openai_beta_header
-            .split(',')
-            .map(str::trim)
-            .any(|value| value == OPENAI_BETA_RESPONSES_WEBSOCKETS)
-    );
-
     server.shutdown().await;
 }
 
@@ -835,7 +804,14 @@ async fn responses_websocket_usage_limit_error_emits_rate_limit_event() {
         }
     });
 
-    let server = start_websocket_server(vec![vec![vec![usage_limit_error]]]).await;
+    let server = start_websocket_server(vec![vec![
+        vec![
+            ev_response_created("resp-prewarm"),
+            ev_completed("resp-prewarm"),
+        ],
+        vec![usage_limit_error],
+    ]])
+    .await;
     let mut builder = test_codex().with_config(|config| {
         config.model_provider.request_max_retries = Some(0);
         config.model_provider.stream_max_retries = Some(0);
@@ -913,7 +889,14 @@ async fn responses_websocket_invalid_request_error_with_status_is_forwarded() {
         }
     });
 
-    let server = start_websocket_server(vec![vec![vec![invalid_request_error]]]).await;
+    let server = start_websocket_server(vec![vec![
+        vec![
+            ev_response_created("resp-prewarm"),
+            ev_completed("resp-prewarm"),
+        ],
+        vec![invalid_request_error],
+    ]])
+    .await;
     let mut builder = test_codex().with_config(|config| {
         config.model_provider.request_max_retries = Some(0);
         config.model_provider.stream_max_retries = Some(0);
@@ -990,14 +973,14 @@ async fn responses_websocket_connection_limit_error_reconnects_and_completes() {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn responses_websocket_appends_on_prefix() {
+async fn responses_websocket_uses_incremental_create_on_prefix() {
     skip_if_no_network!();
 
     let server = start_websocket_server(vec![vec![
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "assistant output"),
-            ev_done(),
+            ev_completed("resp-1"),
         ],
         vec![ev_response_created("resp-2"), ev_completed("resp-2")],
     ]])
@@ -1024,24 +1007,25 @@ async fn responses_websocket_appends_on_prefix() {
     assert_eq!(first["model"].as_str(), Some(MODEL));
     assert_eq!(first["stream"], serde_json::Value::Bool(true));
     assert_eq!(first["input"].as_array().map(Vec::len), Some(1));
-    let expected_append = serde_json::json!({
-        "type": "response.append",
-        "input": serde_json::to_value(&prompt_two.input[2..]).expect("serialize append items"),
-    });
-    assert_eq!(second, expected_append);
+    assert_eq!(second["type"].as_str(), Some("response.create"));
+    assert_eq!(second["previous_response_id"].as_str(), Some("resp-1"));
+    assert_eq!(
+        second["input"],
+        serde_json::to_value(&prompt_two.input[2..]).expect("serialize incremental items")
+    );
 
     server.shutdown().await;
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn responses_websocket_forwards_turn_metadata_on_create_and_append() {
+async fn responses_websocket_forwards_turn_metadata_on_initial_and_incremental_create() {
     skip_if_no_network!();
 
     let server = start_websocket_server(vec![vec![
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "assistant output"),
-            ev_done(),
+            ev_completed("resp-1"),
         ],
         vec![ev_response_created("resp-2"), ev_completed("resp-2")],
     ]])
@@ -1085,7 +1069,8 @@ async fn responses_websocket_forwards_turn_metadata_on_create_and_append() {
         first["client_metadata"]["x-codex-turn-metadata"].as_str(),
         Some(first_turn_metadata)
     );
-    assert_eq!(second["type"].as_str(), Some("response.append"));
+    assert_eq!(second["type"].as_str(), Some("response.create"));
+    assert_eq!(second["previous_response_id"].as_str(), Some("resp-1"));
     assert_eq!(
         second["client_metadata"]["x-codex-turn-metadata"].as_str(),
         Some(enriched_turn_metadata)
@@ -1107,7 +1092,7 @@ async fn responses_websocket_forwards_turn_metadata_on_create_and_append() {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn responses_websocket_creates_on_prefix_when_previous_completion_cannot_append() {
+async fn responses_websocket_uses_previous_response_id_when_prefix_after_completed() {
     skip_if_no_network!();
 
     let server = start_websocket_server(vec![vec![
@@ -1137,9 +1122,10 @@ async fn responses_websocket_creates_on_prefix_when_previous_completion_cannot_a
     let second = connection.get(1).expect("missing request").body_json();
 
     assert_eq!(second["type"].as_str(), Some("response.create"));
+    assert_eq!(second["previous_response_id"].as_str(), Some("resp-1"));
     assert_eq!(
         second["input"],
-        serde_json::to_value(&prompt_two.input).expect("serialize full input")
+        serde_json::to_value(&prompt_two.input[2..]).expect("serialize incremental input")
     );
 
     server.shutdown().await;
@@ -1222,7 +1208,7 @@ async fn responses_websocket_v2_creates_with_previous_response_id_on_prefix() {
         vec![
             ev_response_created("resp-1"),
             ev_assistant_message("msg-1", "assistant output"),
-            ev_done_with_id("resp-1"),
+            ev_completed("resp-1"),
         ],
         vec![ev_response_created("resp-2"), ev_completed("resp-2")],
     ]])
@@ -1409,13 +1395,6 @@ async fn responses_websocket_v2_sets_openai_beta_header() {
             .map(str::trim)
             .any(|value| value == WS_V2_BETA_HEADER_VALUE)
     );
-    assert!(
-        !openai_beta_header
-            .split(',')
-            .map(str::trim)
-            .any(|value| value == OPENAI_BETA_RESPONSES_WEBSOCKETS)
-    );
-
     server.shutdown().await;
 }
 

codex-rs/core/tests/suite/turn_state.rs

@@ -4,7 +4,6 @@ use anyhow::Result;
 use core_test_support::responses::WebSocketConnectionConfig;
 use core_test_support::responses::ev_assistant_message;
 use core_test_support::responses::ev_completed;
-use core_test_support::responses::ev_done;
 use core_test_support::responses::ev_reasoning_item;
 use core_test_support::responses::ev_response_created;
 use core_test_support::responses::ev_shell_command_call;
@@ -100,7 +99,7 @@ async fn websocket_turn_state_persists_within_turn_and_resets_after() -> Result<
                 ev_response_created("resp-1"),
                 ev_reasoning_item("rsn-1", &["thinking"], &[]),
                 ev_shell_command_call(call_id, "echo websocket"),
-                ev_done(),
+                ev_completed("resp-1"),
             ]],
             response_headers: vec![(TURN_STATE_HEADER.to_string(), "ts-1".to_string())],
             accept_delay: None,

codex-rs/core/tests/suite/websocket_fallback.rs

@@ -66,9 +66,9 @@ async fn websocket_fallback_switches_to_http_on_upgrade_required_connect() -> Re
         .filter(|req| req.method == Method::POST && req.url.path().ends_with("/responses"))
         .count();
 
-    // Startup prewarm now only preconnects for v1 (one websocket GET with no request body).
-    // The first turn then attempts websocket once, sees 426, and falls back to HTTP.
-    assert_eq!(websocket_attempts, 2);
+    // The startup prewarm request sees 426 and immediately switches the session to HTTP fallback,
+    // so the first turn goes straight to HTTP with no additional websocket connect attempt.
+    assert_eq!(websocket_attempts, 1);
     assert_eq!(http_attempts, 1);
     assert_eq!(response_mock.requests().len(), 1);
 

codex-rs/feedback/src/lib.rs

@@ -226,6 +226,7 @@ impl CodexLogSnapshot {
         include_logs: bool,
         extra_log_files: &[PathBuf],
         session_source: Option<SessionSource>,
+        logs_override: Option<Vec<u8>>,
     ) -> Result<()> {
         use std::collections::BTreeMap;
         use std::fs;
@@ -310,7 +311,7 @@ impl CodexLogSnapshot {
 
         if include_logs {
             envelope.add_item(EnvelopeItem::Attachment(Attachment {
-                buffer: self.bytes.clone(),
+                buffer: logs_override.unwrap_or_else(|| self.bytes.clone()),
                 filename: String::from("codex-logs.log"),
                 content_type: Some("text/plain".to_string()),
                 ty: None,

codex-rs/protocol/src/prompts/realtime/realtime_end.md

@@ -1,5 +1,3 @@
 Realtime conversation ended.
 
-You are still operating behind an intermediary rather than speaking to the user directly. Use the conversation transcript and current context to decide whether backend help is actually needed, and avoid verbose responses that only add latency.
-
-Subsequent user input may return to typed text rather than transcript-style text. Do not assume recognition errors or missing punctuation once realtime has ended. Resume normal chat behavior.
+Subsequent user input will return to typed text rather than transcript-style text. Do not assume recognition errors or missing punctuation once realtime has ended. Resume normal chat behavior.

codex-rs/protocol/src/prompts/realtime/realtime_start.md

@@ -6,5 +6,4 @@ When invoked, you receive the latest conversation transcript and any relevant mo
 
 When user text is routed from realtime, treat it as a transcript. It may be unpunctuated or contain recognition errors.
 
-- Ask brief clarification questions when needed.
-- Keep responses concise and action-oriented.
+- Keep responses concise and action-oriented. Your updates should help the intermediary respond to the user.

codex-rs/shell-command/src/parse_command.rs

@@ -37,7 +37,26 @@ pub fn parse_command(command: &[String]) -> Vec<ParsedCommand> {
         }
         deduped.push(cmd);
     }
-    deduped
+    if deduped
+        .iter()
+        .any(|cmd| matches!(cmd, ParsedCommand::Unknown { .. }))
+    {
+        vec![single_unknown_for_command(command)]
+    } else {
+        deduped
+    }
+}
+
+fn single_unknown_for_command(command: &[String]) -> ParsedCommand {
+    if let Some((_, shell_command)) = extract_shell_command(command) {
+        ParsedCommand::Unknown {
+            cmd: shell_command.to_string(),
+        }
+    } else {
+        ParsedCommand::Unknown {
+            cmd: shlex_join(command),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -119,7 +138,7 @@ mod tests {
         assert_parsed(
             &vec_str(&["bash", "-lc", inner]),
             vec![ParsedCommand::Unknown {
-                cmd: "git status".to_string(),
+                cmd: inner.to_string(),
             }],
         );
     }
@@ -141,24 +160,9 @@ mod tests {
             "rg --version && node -v && pnpm -v && rg --files | wc -l && rg --files | head -n 40";
         assert_parsed(
             &vec_str(&["bash", "-lc", inner]),
-            vec![
-                // Expect commands in left-to-right execution order
-                ParsedCommand::Search {
-                    cmd: "rg --version".to_string(),
-                    query: None,
-                    path: None,
-                },
-                ParsedCommand::Unknown {
-                    cmd: "node -v".to_string(),
-                },
-                ParsedCommand::Unknown {
-                    cmd: "pnpm -v".to_string(),
-                },
-                ParsedCommand::ListFiles {
-                    cmd: "rg --files".to_string(),
-                    path: None,
-                },
-            ],
+            vec![ParsedCommand::Unknown {
+                cmd: inner.to_string(),
+            }],
         );
     }
 
@@ -218,16 +222,33 @@ mod tests {
         let inner = r#"rg -l QkBindingController presentation/src/main/java | xargs perl -pi -e 's/QkBindingController/QkController/g'"#;
         assert_parsed(
             &vec_str(&["bash", "-lc", inner]),
-            vec![
-                ParsedCommand::Search {
-                    cmd: "rg -l QkBindingController presentation/src/main/java".to_string(),
-                    query: Some("QkBindingController".to_string()),
-                    path: Some("java".to_string()),
-                },
-                ParsedCommand::Unknown {
-                    cmd: "xargs perl -pi -e s/QkBindingController/QkController/g".to_string(),
-                },
-            ],
+            vec![ParsedCommand::Unknown {
+                cmd: inner.to_string(),
+            }],
+        );
+    }
+
+    #[test]
+    fn collapses_plain_pipeline_when_any_stage_is_unknown() {
+        let command = shlex_split_safe(
+            "rg -l QkBindingController presentation/src/main/java | xargs perl -pi -e 's/QkBindingController/QkController/g'",
+        );
+        assert_parsed(
+            &command,
+            vec![ParsedCommand::Unknown {
+                cmd: shlex_join(&command),
+            }],
+        );
+    }
+
+    #[test]
+    fn collapses_pipeline_with_helper_when_later_stage_is_unknown() {
+        let command = shlex_split_safe("rg --files | nl -ba | foo");
+        assert_parsed(
+            &command,
+            vec![ParsedCommand::Unknown {
+                cmd: shlex_join(&command),
+            }],
         );
     }
 
@@ -395,7 +416,7 @@ mod tests {
         assert_parsed(
             &shlex_split_safe("bash -lc 'cd foo && bar'"),
             vec![ParsedCommand::Unknown {
-                cmd: "bar".to_string(),
+                cmd: "cd foo && bar".to_string(),
             }],
         );
     }
@@ -922,85 +943,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn trim_on_semicolon() {
-        assert_parsed(
-            &shlex_split_safe("rg foo ; echo done"),
-            vec![
-                ParsedCommand::Search {
-                    cmd: "rg foo".to_string(),
-                    query: Some("foo".to_string()),
-                    path: None,
-                },
-                ParsedCommand::Unknown {
-                    cmd: "echo done".to_string(),
-                },
-            ],
-        );
-    }
-
-    #[test]
-    fn split_on_or_connector() {
-        // Ensure we split commands on the logical OR operator as well.
-        assert_parsed(
-            &shlex_split_safe("rg foo || echo done"),
-            vec![
-                ParsedCommand::Search {
-                    cmd: "rg foo".to_string(),
-                    query: Some("foo".to_string()),
-                    path: None,
-                },
-                ParsedCommand::Unknown {
-                    cmd: "echo done".to_string(),
-                },
-            ],
-        );
-    }
-
-    #[test]
-    fn parses_mixed_sequence_with_pipes_semicolons_and_or() {
-        // Provided long command sequence combining sequencing, pipelines, and ORs.
-        let inner = "pwd; ls -la; rg --files -g '!target' | wc -l; rg -n '^\\[workspace\\]' -n Cargo.toml || true; rg -n '^\\[package\\]' -n */Cargo.toml || true; cargo --version; rustc --version; cargo clippy --workspace --all-targets --all-features -q";
-        let args = vec_str(&["bash", "-lc", inner]);
-
-        let expected = vec![
-            ParsedCommand::Unknown {
-                cmd: "pwd".to_string(),
-            },
-            ParsedCommand::ListFiles {
-                cmd: shlex_join(&shlex_split_safe("ls -la")),
-                path: None,
-            },
-            ParsedCommand::ListFiles {
-                cmd: shlex_join(&shlex_split_safe("rg --files -g '!target'")),
-                path: None,
-            },
-            ParsedCommand::Search {
-                cmd: shlex_join(&shlex_split_safe("rg -n '^\\[workspace\\]' -n Cargo.toml")),
-                query: Some("^\\[workspace\\]".to_string()),
-                path: Some("Cargo.toml".to_string()),
-            },
-            ParsedCommand::Search {
-                cmd: shlex_join(&shlex_split_safe("rg -n '^\\[package\\]' -n */Cargo.toml")),
-                query: Some("^\\[package\\]".to_string()),
-                path: Some("Cargo.toml".to_string()),
-            },
-            ParsedCommand::Unknown {
-                cmd: shlex_join(&shlex_split_safe("cargo --version")),
-            },
-            ParsedCommand::Unknown {
-                cmd: shlex_join(&shlex_split_safe("rustc --version")),
-            },
-            ParsedCommand::Unknown {
-                cmd: shlex_join(&shlex_split_safe(
-                    "cargo clippy --workspace --all-targets --all-features -q",
-                )),
-            },
-        ];
-
-        assert_parsed(&args, expected);
-    }
-
     #[test]
     fn strips_true_in_sequence() {
         // `true` should be dropped from parsed sequences
@@ -1926,6 +1868,7 @@ fn parse_shell_lc_commands(original: &[String]) -> Option<Vec<ParsedCommand>> {
             };
             commands.push(parsed);
         }
+
         if commands.len() > 1 {
             commands.retain(|pc| !matches!(pc, ParsedCommand::Unknown { cmd } if cmd == "true"));
             // Apply the same simplifications used for non-bash parsing, e.g., drop leading `cd`.

codex-rs/shell-escalation/src/lib.rs

@@ -14,6 +14,8 @@ pub use unix::EscalationPermissions;
 #[cfg(unix)]
 pub use unix::EscalationPolicy;
 #[cfg(unix)]
+pub use unix::EscalationSession;
+#[cfg(unix)]
 pub use unix::ExecParams;
 #[cfg(unix)]
 pub use unix::ExecResult;

codex-rs/shell-escalation/src/unix/escalate_server.rs

@@ -7,7 +7,9 @@ use std::time::Duration;
 
 use anyhow::Context as _;
 use codex_utils_absolute_path::AbsolutePathBuf;
+use socket2::Socket;
 use tokio::process::Command;
+use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 
 use crate::unix::escalate_protocol::ESCALATE_SOCKET_ENV_VAR;
@@ -82,6 +84,25 @@ pub struct PreparedExec {
     pub arg0: Option<String>,
 }
 
+#[derive(Debug)]
+pub struct EscalationSession {
+    env: HashMap<String, String>,
+    task: JoinHandle<anyhow::Result<()>>,
+    _client_socket: Socket,
+}
+
+impl EscalationSession {
+    pub fn env(&self) -> &HashMap<String, String> {
+        &self.env
+    }
+}
+
+impl Drop for EscalationSession {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
 pub struct EscalateServer {
     bash_path: PathBuf,
     execve_wrapper: PathBuf,
@@ -106,16 +127,42 @@ impl EscalateServer {
         cancel_rx: CancellationToken,
         command_executor: Arc<dyn ShellCommandExecutor>,
     ) -> anyhow::Result<ExecResult> {
+        let session = self.start_session(Arc::clone(&command_executor))?;
+        let command = vec![
+            self.bash_path.to_string_lossy().to_string(),
+            if params.login == Some(false) {
+                "-c".to_string()
+            } else {
+                "-lc".to_string()
+            },
+            params.command,
+        ];
+        let workdir = AbsolutePathBuf::try_from(params.workdir)?;
+        let result = command_executor
+            .run(
+                command,
+                workdir.to_path_buf(),
+                session.env().clone(),
+                cancel_rx,
+            )
+            .await?;
+        Ok(result)
+    }
+
+    pub fn start_session(
+        &self,
+        command_executor: Arc<dyn ShellCommandExecutor>,
+    ) -> anyhow::Result<EscalationSession> {
         let (escalate_server, escalate_client) = AsyncDatagramSocket::pair()?;
         let client_socket = escalate_client.into_inner();
         // Only the client endpoint should cross exec into the wrapper process.
         client_socket.set_cloexec(false)?;
-        let escalate_task = tokio::spawn(escalate_task(
+        let task = tokio::spawn(escalate_task(
             escalate_server,
             Arc::clone(&self.policy),
             Arc::clone(&command_executor),
         ));
-        let mut env = std::env::vars().collect::<HashMap<String, String>>();
+        let mut env = HashMap::new();
         env.insert(
             ESCALATE_SOCKET_ENV_VAR.to_string(),
             client_socket.as_raw_fd().to_string(),
@@ -128,22 +175,11 @@ impl EscalateServer {
             LEGACY_BASH_EXEC_WRAPPER_ENV_VAR.to_string(),
             self.execve_wrapper.to_string_lossy().to_string(),
         );
-
-        let command = vec![
-            self.bash_path.to_string_lossy().to_string(),
-            if params.login == Some(false) {
-                "-c".to_string()
-            } else {
-                "-lc".to_string()
-            },
-            params.command,
-        ];
-        let workdir = AbsolutePathBuf::try_from(params.workdir)?;
-        let result = command_executor
-            .run(command, workdir.to_path_buf(), env, cancel_rx)
-            .await?;
-        escalate_task.abort();
-        Ok(result)
+        Ok(EscalationSession {
+            env,
+            task,
+            _client_socket: client_socket,
+        })
     }
 }
 
@@ -390,6 +426,35 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn start_session_exposes_wrapper_env_overlay() -> anyhow::Result<()> {
+        let execve_wrapper = PathBuf::from("/tmp/codex-execve-wrapper");
+        let execve_wrapper_str = execve_wrapper.to_string_lossy().to_string();
+        let server = EscalateServer::new(
+            PathBuf::from("/bin/bash"),
+            execve_wrapper.clone(),
+            DeterministicEscalationPolicy {
+                decision: EscalationDecision::run(),
+            },
+        );
+
+        let session = server.start_session(Arc::new(ForwardingShellCommandExecutor))?;
+        let env = session.env();
+        assert_eq!(env.get(EXEC_WRAPPER_ENV_VAR), Some(&execve_wrapper_str));
+        assert_eq!(
+            env.get(LEGACY_BASH_EXEC_WRAPPER_ENV_VAR),
+            Some(&execve_wrapper_str)
+        );
+        let socket_fd = env
+            .get(ESCALATE_SOCKET_ENV_VAR)
+            .expect("session should export shell escalation socket");
+        let socket_fd = socket_fd.parse::<i32>()?;
+        assert!(socket_fd >= 0);
+        assert_ne!(unsafe { libc::fcntl(socket_fd, libc::F_GETFD) }, -1);
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn handle_escalate_session_respects_run_in_sandbox_decision() -> anyhow::Result<()> {
         let (server, client) = AsyncSocket::pair()?;

codex-rs/shell-escalation/src/unix/mod.rs

@@ -66,6 +66,7 @@ pub use self::escalate_protocol::EscalateAction;
 pub use self::escalate_protocol::EscalationDecision;
 pub use self::escalate_protocol::EscalationExecution;
 pub use self::escalate_server::EscalateServer;
+pub use self::escalate_server::EscalationSession;
 pub use self::escalate_server::ExecParams;
 pub use self::escalate_server::ExecResult;
 pub use self::escalate_server::PreparedExec;

codex-rs/shell-escalation/src/unix/stopwatch.rs

@@ -9,7 +9,7 @@ use tokio_util::sync::CancellationToken;
 
 #[derive(Clone, Debug)]
 pub struct Stopwatch {
-    limit: Duration,
+    limit: Option<Duration>,
     inner: Arc<Mutex<StopwatchState>>,
     notify: Arc<Notify>,
 }
@@ -30,13 +30,27 @@ impl Stopwatch {
                 active_pauses: 0,
             })),
             notify: Arc::new(Notify::new()),
-            limit,
+            limit: Some(limit),
+        }
+    }
+
+    pub fn unlimited() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(StopwatchState {
+                elapsed: Duration::ZERO,
+                running_since: Some(Instant::now()),
+                active_pauses: 0,
+            })),
+            notify: Arc::new(Notify::new()),
+            limit: None,
         }
     }
 
     pub fn cancellation_token(&self) -> CancellationToken {
-        let limit = self.limit;
         let token = CancellationToken::new();
+        let Some(limit) = self.limit else {
+            return token;
+        };
         let cancel = token.clone();
         let inner = Arc::clone(&self.inner);
         let notify = Arc::clone(&self.notify);
@@ -208,4 +222,16 @@ mod tests {
         // Now the stopwatch should resume and hit the limit shortly after.
         token.cancelled().await;
     }
+
+    #[tokio::test]
+    async fn unlimited_stopwatch_never_cancels() {
+        let stopwatch = Stopwatch::unlimited();
+        let token = stopwatch.cancellation_token();
+
+        assert!(
+            timeout(Duration::from_millis(30), token.cancelled())
+                .await
+                .is_err()
+        );
+    }
 }

codex-rs/state/src/log_db.rs

@@ -26,6 +26,7 @@ use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 
 use tokio::sync::mpsc;
+use tokio::sync::oneshot;
 use tracing::Event;
 use tracing::field::Field;
 use tracing::field::Visit;
@@ -45,7 +46,7 @@ const LOG_FLUSH_INTERVAL: Duration = Duration::from_millis(250);
 const LOG_RETENTION_DAYS: i64 = 90;
 
 pub struct LogDbLayer {
-    sender: mpsc::Sender<LogEntry>,
+    sender: mpsc::Sender<LogDbCommand>,
     process_uuid: String,
 }
 
@@ -61,6 +62,24 @@ pub fn start(state_db: std::sync::Arc<StateRuntime>) -> LogDbLayer {
     }
 }
 
+impl Clone for LogDbLayer {
+    fn clone(&self) -> Self {
+        Self {
+            sender: self.sender.clone(),
+            process_uuid: self.process_uuid.clone(),
+        }
+    }
+}
+
+impl LogDbLayer {
+    pub async fn flush(&self) {
+        let (tx, rx) = oneshot::channel();
+        if self.sender.send(LogDbCommand::Flush(tx)).await.is_ok() {
+            let _ = rx.await;
+        }
+    }
+}
+
 impl<S> Layer<S> for LogDbLayer
 where
     S: tracing::Subscriber + for<'a> LookupSpan<'a>,
@@ -131,10 +150,15 @@ where
             line: metadata.line().map(|line| line as i64),
         };
 
-        let _ = self.sender.try_send(entry);
+        let _ = self.sender.try_send(LogDbCommand::Entry(entry));
     }
 }
 
+enum LogDbCommand {
+    Entry(LogEntry),
+    Flush(oneshot::Sender<()>),
+}
+
 #[derive(Clone, Debug, Default)]
 struct SpanLogContext {
     thread_id: Option<String>,
@@ -215,20 +239,24 @@ fn current_process_log_uuid() -> &'static str {
 
 async fn run_inserter(
     state_db: std::sync::Arc<StateRuntime>,
-    mut receiver: mpsc::Receiver<LogEntry>,
+    mut receiver: mpsc::Receiver<LogDbCommand>,
 ) {
     let mut buffer = Vec::with_capacity(LOG_BATCH_SIZE);
     let mut ticker = tokio::time::interval(LOG_FLUSH_INTERVAL);
     loop {
         tokio::select! {
-            maybe_entry = receiver.recv() => {
-                match maybe_entry {
-                    Some(entry) => {
+            maybe_command = receiver.recv() => {
+                match maybe_command {
+                    Some(LogDbCommand::Entry(entry)) => {
                         buffer.push(entry);
                         if buffer.len() >= LOG_BATCH_SIZE {
                             flush(&state_db, &mut buffer).await;
                         }
                     }
+                    Some(LogDbCommand::Flush(reply)) => {
+                        flush(&state_db, &mut buffer).await;
+                        let _ = reply.send(());
+                    }
                     None => {
                         flush(&state_db, &mut buffer).await;
                         break;
@@ -304,3 +332,150 @@ impl Visit for MessageVisitor {
         self.record_field(field, format!("{value:?}"));
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::io;
+    use std::sync::Arc;
+    use std::sync::Mutex;
+    use std::time::Duration;
+
+    use tokio::time::Instant;
+    use tracing_subscriber::filter::Targets;
+    use tracing_subscriber::fmt::writer::MakeWriter;
+    use tracing_subscriber::layer::SubscriberExt;
+    use tracing_subscriber::util::SubscriberInitExt;
+
+    use super::*;
+
+    #[derive(Clone, Default)]
+    struct SharedWriter {
+        bytes: Arc<Mutex<Vec<u8>>>,
+    }
+
+    impl SharedWriter {
+        fn snapshot(&self) -> String {
+            String::from_utf8(self.bytes.lock().expect("writer mutex poisoned").clone())
+                .expect("valid utf-8")
+        }
+    }
+
+    struct SharedWriterGuard {
+        bytes: Arc<Mutex<Vec<u8>>>,
+    }
+
+    impl<'a> MakeWriter<'a> for SharedWriter {
+        type Writer = SharedWriterGuard;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            SharedWriterGuard {
+                bytes: Arc::clone(&self.bytes),
+            }
+        }
+    }
+
+    impl io::Write for SharedWriterGuard {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.bytes
+                .lock()
+                .expect("writer mutex poisoned")
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn sqlite_feedback_logs_match_feedback_formatter_shape() {
+        let codex_home =
+            std::env::temp_dir().join(format!("codex-state-log-db-{}", Uuid::new_v4()));
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+        let writer = SharedWriter::default();
+
+        let subscriber = tracing_subscriber::registry()
+            .with(
+                tracing_subscriber::fmt::layer()
+                    .with_writer(writer.clone())
+                    .without_time()
+                    .with_ansi(false)
+                    .with_target(false)
+                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
+            )
+            .with(
+                start(runtime.clone())
+                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
+            );
+        let guard = subscriber.set_default();
+
+        tracing::trace!("threadless-before");
+        tracing::info_span!("feedback-thread", thread_id = "thread-1").in_scope(|| {
+            tracing::info!("thread-scoped");
+        });
+        tracing::debug!("threadless-after");
+
+        drop(guard);
+
+        // TODO(ccunningham): Store enough span metadata in SQLite to reproduce span
+        // prefixes like `feedback-thread{thread_id="thread-1"}:` in feedback exports.
+        let feedback_logs = writer
+            .snapshot()
+            .replace("feedback-thread{thread_id=\"thread-1\"}: ", "");
+        let deadline = Instant::now() + Duration::from_secs(2);
+        loop {
+            let sqlite_logs = String::from_utf8(
+                runtime
+                    .query_feedback_logs("thread-1")
+                    .await
+                    .expect("query feedback logs"),
+            )
+            .expect("valid utf-8");
+            if sqlite_logs == feedback_logs {
+                break;
+            }
+            assert!(
+                Instant::now() < deadline,
+                "sqlite feedback logs did not match feedback formatter output before timeout\nsqlite:\n{sqlite_logs}\nfeedback:\n{feedback_logs}"
+            );
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
+    #[tokio::test]
+    async fn flush_persists_logs_for_query() {
+        let codex_home =
+            std::env::temp_dir().join(format!("codex-state-log-db-{}", Uuid::new_v4()));
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+        let layer = start(runtime.clone());
+
+        let guard = tracing_subscriber::registry()
+            .with(
+                layer
+                    .clone()
+                    .with_filter(Targets::new().with_default(tracing::Level::TRACE)),
+            )
+            .set_default();
+
+        tracing::info!("buffered-log");
+
+        layer.flush().await;
+        drop(guard);
+
+        let after_flush = runtime
+            .query_logs(&crate::LogQuery::default())
+            .await
+            .expect("query logs after flush");
+        assert_eq!(after_flush.len(), 1);
+        assert_eq!(after_flush[0].message.as_deref(), Some("buffered-log"));
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+}

codex-rs/state/src/runtime/logs.rs

@@ -285,6 +285,67 @@ WHERE id IN (
         Ok(rows)
     }
 
+    /// Query per-thread feedback logs, capped to the per-thread SQLite retention budget.
+    pub async fn query_feedback_logs(&self, thread_id: &str) -> anyhow::Result<Vec<u8>> {
+        let max_bytes = LOG_PARTITION_SIZE_LIMIT_BYTES;
+        let lines = sqlx::query_scalar::<_, String>(
+            r#"
+WITH latest_process AS (
+    SELECT process_uuid
+    FROM logs
+    WHERE thread_id = ? AND process_uuid IS NOT NULL
+    ORDER BY ts DESC, ts_nanos DESC, id DESC
+    LIMIT 1
+),
+feedback_logs AS (
+    SELECT
+        printf('%5s %s', level, message) || CASE
+            WHEN substr(message, -1, 1) = char(10) THEN ''
+            ELSE char(10)
+        END AS line,
+        length(CAST(
+            printf('%5s %s', level, message) || CASE
+                WHEN substr(message, -1, 1) = char(10) THEN ''
+                ELSE char(10)
+            END AS BLOB
+        )) AS line_bytes,
+        ts,
+        ts_nanos,
+        id
+    FROM logs
+    WHERE message IS NOT NULL AND (
+        thread_id = ?
+        OR (
+            thread_id IS NULL
+            AND process_uuid IN (SELECT process_uuid FROM latest_process)
+        )
+    )
+)
+SELECT line
+FROM (
+    SELECT
+        line,
+        ts,
+        ts_nanos,
+        id,
+        SUM(line_bytes) OVER (
+            ORDER BY ts DESC, ts_nanos DESC, id DESC
+        ) AS cumulative_bytes
+    FROM feedback_logs
+)
+WHERE cumulative_bytes <= ?
+ORDER BY ts ASC, ts_nanos ASC, id ASC
+"#,
+        )
+        .bind(thread_id)
+        .bind(thread_id)
+        .bind(max_bytes)
+        .fetch_all(self.pool.as_ref())
+        .await?;
+
+        Ok(lines.concat().into_bytes())
+    }
+
     /// Return the max log id matching optional filters.
     pub async fn max_log_id(&self, query: &LogQuery) -> anyhow::Result<i64> {
         let mut builder =
@@ -712,4 +773,334 @@ mod tests {
 
         let _ = tokio::fs::remove_dir_all(codex_home).await;
     }
+
+    #[tokio::test]
+    async fn query_feedback_logs_returns_newest_lines_within_limit_in_order() {
+        let codex_home = unique_temp_dir();
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+
+        runtime
+            .insert_logs(&[
+                LogEntry {
+                    ts: 1,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("alpha".to_string()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 2,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("bravo".to_string()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 3,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("charlie".to_string()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+            ])
+            .await
+            .expect("insert test logs");
+
+        let bytes = runtime
+            .query_feedback_logs("thread-1")
+            .await
+            .expect("query feedback logs");
+
+        assert_eq!(
+            String::from_utf8(bytes).expect("valid utf-8"),
+            " INFO alpha\n INFO bravo\n INFO charlie\n"
+        );
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
+    #[tokio::test]
+    async fn query_feedback_logs_excludes_oversized_newest_row() {
+        let codex_home = unique_temp_dir();
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+        let eleven_mebibytes = "z".repeat(11 * 1024 * 1024);
+
+        runtime
+            .insert_logs(&[
+                LogEntry {
+                    ts: 1,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("small".to_string()),
+                    thread_id: Some("thread-oversized".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 2,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some(eleven_mebibytes),
+                    thread_id: Some("thread-oversized".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+            ])
+            .await
+            .expect("insert test logs");
+
+        let bytes = runtime
+            .query_feedback_logs("thread-oversized")
+            .await
+            .expect("query feedback logs");
+
+        assert_eq!(bytes, Vec::<u8>::new());
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
+    #[tokio::test]
+    async fn query_feedback_logs_includes_threadless_rows_from_same_process() {
+        let codex_home = unique_temp_dir();
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+
+        runtime
+            .insert_logs(&[
+                LogEntry {
+                    ts: 1,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("threadless-before".to_string()),
+                    thread_id: None,
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 2,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("thread-scoped".to_string()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 3,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("threadless-after".to_string()),
+                    thread_id: None,
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 4,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("other-process-threadless".to_string()),
+                    thread_id: None,
+                    process_uuid: Some("proc-2".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+            ])
+            .await
+            .expect("insert test logs");
+
+        let bytes = runtime
+            .query_feedback_logs("thread-1")
+            .await
+            .expect("query feedback logs");
+
+        assert_eq!(
+            String::from_utf8(bytes).expect("valid utf-8"),
+            " INFO threadless-before\n INFO thread-scoped\n INFO threadless-after\n"
+        );
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
+    #[tokio::test]
+    async fn query_feedback_logs_excludes_threadless_rows_from_prior_processes() {
+        let codex_home = unique_temp_dir();
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+
+        runtime
+            .insert_logs(&[
+                LogEntry {
+                    ts: 1,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("old-process-threadless".to_string()),
+                    thread_id: None,
+                    process_uuid: Some("proc-old".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 2,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("old-process-thread".to_string()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-old".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 3,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("new-process-thread".to_string()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-new".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 4,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some("new-process-threadless".to_string()),
+                    thread_id: None,
+                    process_uuid: Some("proc-new".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+            ])
+            .await
+            .expect("insert test logs");
+
+        let bytes = runtime
+            .query_feedback_logs("thread-1")
+            .await
+            .expect("query feedback logs");
+
+        assert_eq!(
+            String::from_utf8(bytes).expect("valid utf-8"),
+            " INFO old-process-thread\n INFO new-process-thread\n INFO new-process-threadless\n"
+        );
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
+    #[tokio::test]
+    async fn query_feedback_logs_keeps_newest_suffix_across_thread_and_threadless_logs() {
+        let codex_home = unique_temp_dir();
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
+            .await
+            .expect("initialize runtime");
+        let thread_marker = "thread-scoped-oldest";
+        let threadless_older_marker = "threadless-older";
+        let threadless_newer_marker = "threadless-newer";
+        let five_mebibytes = format!("{threadless_older_marker} {}", "a".repeat(5 * 1024 * 1024));
+        let four_and_half_mebibytes = format!(
+            "{threadless_newer_marker} {}",
+            "b".repeat((9 * 1024 * 1024) / 2)
+        );
+        let one_mebibyte = format!("{thread_marker} {}", "c".repeat(1024 * 1024));
+
+        runtime
+            .insert_logs(&[
+                LogEntry {
+                    ts: 1,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some(one_mebibyte.clone()),
+                    thread_id: Some("thread-1".to_string()),
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 2,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some(five_mebibytes),
+                    thread_id: None,
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+                LogEntry {
+                    ts: 3,
+                    ts_nanos: 0,
+                    level: "INFO".to_string(),
+                    target: "cli".to_string(),
+                    message: Some(four_and_half_mebibytes),
+                    thread_id: None,
+                    process_uuid: Some("proc-1".to_string()),
+                    file: None,
+                    line: None,
+                    module_path: None,
+                },
+            ])
+            .await
+            .expect("insert test logs");
+
+        let bytes = runtime
+            .query_feedback_logs("thread-1")
+            .await
+            .expect("query feedback logs");
+        let logs = String::from_utf8(bytes).expect("valid utf-8");
+
+        assert!(!logs.contains(thread_marker));
+        assert!(logs.contains(threadless_older_marker));
+        assert!(logs.contains(threadless_newer_marker));
+        assert_eq!(logs.matches('\n').count(), 2);
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
 }

codex-rs/tui/src/bottom_pane/feedback_view.rs

@@ -102,6 +102,7 @@ impl FeedbackNoteView {
             self.include_logs,
             &log_file_paths,
             Some(SessionSource::Cli),
+            None,
         );
 
         match result {

scripts/install/install.ps1

@@ -0,0 +1,196 @@
+param(
+    [Parameter(Position=0)]
+    [string]$Version = "latest"
+)
+
+Set-StrictMode -Version Latest
+$ErrorActionPreference = "Stop"
+$ProgressPreference = "SilentlyContinue"
+
+function Write-Step {
+    param(
+        [string]$Message
+    )
+
+    Write-Host "==> $Message"
+}
+
+function Normalize-Version {
+    param(
+        [string]$RawVersion
+    )
+
+    if ([string]::IsNullOrWhiteSpace($RawVersion) -or $RawVersion -eq "latest") {
+        return "latest"
+    }
+
+    if ($RawVersion.StartsWith("rust-v")) {
+        return $RawVersion.Substring(6)
+    }
+
+    if ($RawVersion.StartsWith("v")) {
+        return $RawVersion.Substring(1)
+    }
+
+    return $RawVersion
+}
+
+function Get-ReleaseUrl {
+    param(
+        [string]$AssetName,
+        [string]$ResolvedVersion
+    )
+
+    return "https://github.com/openai/codex/releases/download/rust-v$ResolvedVersion/$AssetName"
+}
+
+function Path-Contains {
+    param(
+        [string]$PathValue,
+        [string]$Entry
+    )
+
+    if ([string]::IsNullOrWhiteSpace($PathValue)) {
+        return $false
+    }
+
+    $needle = $Entry.TrimEnd("\")
+    foreach ($segment in $PathValue.Split(";", [System.StringSplitOptions]::RemoveEmptyEntries)) {
+        if ($segment.TrimEnd("\") -ieq $needle) {
+            return $true
+        }
+    }
+
+    return $false
+}
+
+function Resolve-Version {
+    $normalizedVersion = Normalize-Version -RawVersion $Version
+    if ($normalizedVersion -ne "latest") {
+        return $normalizedVersion
+    }
+
+    $release = Invoke-RestMethod -Uri "https://api.github.com/repos/openai/codex/releases/latest"
+    if (-not $release.tag_name) {
+        Write-Error "Failed to resolve the latest Codex release version."
+        exit 1
+    }
+
+    return (Normalize-Version -RawVersion $release.tag_name)
+}
+
+if ($env:OS -ne "Windows_NT") {
+    Write-Error "install.ps1 supports Windows only. Use install.sh on macOS or Linux."
+    exit 1
+}
+
+if (-not [Environment]::Is64BitOperatingSystem) {
+    Write-Error "Codex requires a 64-bit version of Windows."
+    exit 1
+}
+
+$architecture = [System.Runtime.InteropServices.RuntimeInformation]::OSArchitecture
+$target = $null
+$platformLabel = $null
+$npmTag = $null
+switch ($architecture) {
+    "Arm64" {
+        $target = "aarch64-pc-windows-msvc"
+        $platformLabel = "Windows (ARM64)"
+        $npmTag = "win32-arm64"
+    }
+    "X64" {
+        $target = "x86_64-pc-windows-msvc"
+        $platformLabel = "Windows (x64)"
+        $npmTag = "win32-x64"
+    }
+    default {
+        Write-Error "Unsupported architecture: $architecture"
+        exit 1
+    }
+}
+
+if ([string]::IsNullOrWhiteSpace($env:CODEX_INSTALL_DIR)) {
+    $installDir = Join-Path $env:LOCALAPPDATA "Programs\OpenAI\Codex\bin"
+} else {
+    $installDir = $env:CODEX_INSTALL_DIR
+}
+
+$codexPath = Join-Path $installDir "codex.exe"
+$installMode = if (Test-Path $codexPath) { "Updating" } else { "Installing" }
+
+Write-Step "$installMode Codex CLI"
+Write-Step "Detected platform: $platformLabel"
+
+New-Item -ItemType Directory -Force -Path $installDir | Out-Null
+
+$resolvedVersion = Resolve-Version
+Write-Step "Resolved version: $resolvedVersion"
+$packageAsset = "codex-npm-$npmTag-$resolvedVersion.tgz"
+
+$tempDir = Join-Path ([System.IO.Path]::GetTempPath()) ("codex-install-" + [System.Guid]::NewGuid().ToString("N"))
+New-Item -ItemType Directory -Force -Path $tempDir | Out-Null
+
+try {
+    $archivePath = Join-Path $tempDir $packageAsset
+    $extractDir = Join-Path $tempDir "extract"
+    $url = Get-ReleaseUrl -AssetName $packageAsset -ResolvedVersion $resolvedVersion
+
+    Write-Step "Downloading Codex CLI"
+    Invoke-WebRequest -Uri $url -OutFile $archivePath
+
+    New-Item -ItemType Directory -Force -Path $extractDir | Out-Null
+    tar -xzf $archivePath -C $extractDir
+
+    $vendorRoot = Join-Path $extractDir "package/vendor/$target"
+    Write-Step "Installing to $installDir"
+    $copyMap = @{
+        "codex/codex.exe" = "codex.exe"
+        "codex/codex-command-runner.exe" = "codex-command-runner.exe"
+        "codex/codex-windows-sandbox-setup.exe" = "codex-windows-sandbox-setup.exe"
+        "path/rg.exe" = "rg.exe"
+    }
+
+    foreach ($relativeSource in $copyMap.Keys) {
+        $sourcePath = Join-Path $vendorRoot $relativeSource
+        $destinationPath = Join-Path $installDir $copyMap[$relativeSource]
+        Move-Item -Force $sourcePath $destinationPath
+    }
+} finally {
+    Remove-Item -Recurse -Force $tempDir -ErrorAction SilentlyContinue
+}
+
+$userPath = [Environment]::GetEnvironmentVariable("Path", "User")
+$pathNeedsNewShell = $false
+if (-not (Path-Contains -PathValue $userPath -Entry $installDir)) {
+    if ([string]::IsNullOrWhiteSpace($userPath)) {
+        $newUserPath = $installDir
+    } else {
+        $newUserPath = "$installDir;$userPath"
+    }
+
+    [Environment]::SetEnvironmentVariable("Path", $newUserPath, "User")
+    if (-not (Path-Contains -PathValue $env:Path -Entry $installDir)) {
+        if ([string]::IsNullOrWhiteSpace($env:Path)) {
+            $env:Path = $installDir
+        } else {
+            $env:Path = "$installDir;$env:Path"
+        }
+    }
+    Write-Step "PATH updated for future PowerShell sessions."
+    $pathNeedsNewShell = $true
+} elseif (Path-Contains -PathValue $env:Path -Entry $installDir) {
+    Write-Step "$installDir is already on PATH."
+} else {
+    Write-Step "PATH is already configured for future PowerShell sessions."
+    $pathNeedsNewShell = $true
+}
+
+if ($pathNeedsNewShell) {
+    Write-Step ('Run now: $env:Path = "{0};$env:Path"; codex' -f $installDir)
+    Write-Step "Or open a new PowerShell window and run: codex"
+} else {
+    Write-Step "Run: codex"
+}
+
+Write-Host "Codex CLI $resolvedVersion installed successfully."