exec: handle typed turn completion for fanout jobs

# Summary

This PR introduces the Windows sandbox runner IPC foundation that later
unified_exec work will build on.

The key point is that this is intentionally infrastructure-only. The new
IPC transport, runner plumbing, and ConPTY helpers are added here, but
the active elevated Windows sandbox path still uses the existing
request-file bootstrap. In other words, this change prepares the
transport and module layout we need for unified_exec without switching
production behavior over yet.

Part of this PR is also a source-layout cleanup: some Windows sandbox
files are moved into more explicit `elevated/`, `conpty/`, and shared
locations so it is clearer which code is for the elevated sandbox flow,
which code is legacy/direct-spawn behavior, and which helpers are shared
between them. That reorganization is intentional in this first PR so
later behavioral changes do not also have to carry a large amount of
file-move churn.

# Why This Is Needed For unified_exec

Windows elevated sandboxed unified_exec needs a long-lived,
bidirectional control channel between the CLI and a helper process
running under the sandbox user. That channel has to support:

- starting a process and reporting structured spawn success/failure
- streaming stdout/stderr back incrementally
- forwarding stdin over time
- terminating or polling a long-lived process
- supporting both pipe-backed and PTY-backed sessions

The existing elevated one-shot path is built around a request-file
bootstrap and does not provide those primitives cleanly. Before we can
turn on Windows sandbox unified_exec, we need the underlying runner
protocol and transport layer that can carry those lifecycle events and
streams.

# Why Windows Needs More Machinery Than Linux Or macOS

Linux and macOS can generally build unified_exec on top of the existing
sandbox/process model: the parent can spawn the child directly, retain
normal ownership of stdio or PTY handles, and manage the lifetime of the
sandboxed process without introducing a second control process.

Windows elevated sandboxing is different. To run inside the sandbox
boundary, we cross into a different user/security context and then need
to manage a long-lived process from outside that boundary. That means we
need an explicit helper process plus an IPC transport to carry spawn,
stdin, output, and exit events back and forth. The extra code here is
mostly that missing Windows sandbox infrastructure, not a conceptual
difference in unified_exec itself.

# What This PR Adds

- the framed IPC message types and transport helpers for parent <->
runner communication
- the renamed Windows command runner with both the existing request-file
bootstrap and the dormant IPC bootstrap
- named-pipe helpers for the elevated runner path
- ConPTY helpers and process-thread attribute plumbing needed for
PTY-backed sessions
- shared sandbox/process helpers that later PRs will reuse when
switching live execution paths over
- early file/module moves so later PRs can focus on behavior rather than
layout churn

# What This PR Does Not Yet Do

- it does not switch the active elevated one-shot path over to IPC yet
- it does not enable Windows sandbox unified_exec yet
- it does not remove the existing request-file bootstrap yet

So while this code compiles and the new path has basic validation, it is
not yet the exercised production path. That is intentional for this
first PR: the goal here is to land the transport and runner foundation
cleanly before later PRs start routing real command execution through
it.

# Follow-Ups

Planned follow-up PRs will:

1. switch elevated one-shot Windows sandbox execution to the new runner
IPC path
2. layer Windows sandbox unified_exec sessions on top of the same
transport
3. remove the legacy request-file path once the IPC-based path is live

# Validation

- `cargo build -p codex-windows-sandbox`

codex-rs/Cargo.lock

@@ -2846,6 +2846,7 @@ dependencies = [
  "chrono",
  "codex-protocol",
  "codex-utils-absolute-path",
+ "codex-utils-pty",
  "codex-utils-string",
  "dirs-next",
  "dunce",
@@ -2854,6 +2855,7 @@ dependencies = [
  "serde",
  "serde_json",
  "tempfile",
+ "tokio",
  "windows 0.58.0",
  "windows-sys 0.52.0",
  "winres",

codex-rs/core/src/connectors.rs

@@ -469,7 +469,7 @@ pub fn connector_display_label(connector: &AppInfo) -> String {
 }
 
 pub fn connector_mention_slug(connector: &AppInfo) -> String {
-    sanitize_name(&connector_display_label(connector))
+    sanitize_slug(&connector_display_label(connector))
 }
 
 pub(crate) fn accessible_connectors_from_mcp_tools(
@@ -918,11 +918,15 @@ fn normalize_connector_value(value: Option<&str>) -> Option<String> {
 }
 
 pub fn connector_install_url(name: &str, connector_id: &str) -> String {
-    let slug = sanitize_name(name);
+    let slug = sanitize_slug(name);
     format!("https://chatgpt.com/apps/{slug}/{connector_id}")
 }
 
 pub fn sanitize_name(name: &str) -> String {
+    sanitize_slug(name).replace("-", "_")
+}
+
+fn sanitize_slug(name: &str) -> String {
     let mut normalized = String::with_capacity(name.len());
     for character in name.chars() {
         if character.is_ascii_alphanumeric() {

codex-rs/core/src/contextual_user_message.rs

@@ -103,6 +103,21 @@ pub(crate) fn is_contextual_user_fragment(content_item: &ContentItem) -> bool {
         .any(|definition| definition.matches_text(text))
 }
 
+/// Returns whether a contextual user fragment should be omitted from memory
+/// stage-1 inputs.
+///
+/// We exclude injected `AGENTS.md` instructions and skill payloads because
+/// they are prompt scaffolding rather than conversation content, so they do
+/// not improve the resulting memory. We keep environment context and
+/// subagent notifications because they can carry useful execution context or
+/// subtask outcomes that should remain visible to memory generation.
+pub(crate) fn is_memory_excluded_contextual_user_fragment(content_item: &ContentItem) -> bool {
+    let ContentItem::InputText { text } = content_item else {
+        return false;
+    };
+    AGENTS_MD_FRAGMENT.matches_text(text) || SKILL_FRAGMENT.matches_text(text)
+}
+
 #[cfg(test)]
 #[path = "contextual_user_message_tests.rs"]
 mod tests;

codex-rs/core/src/contextual_user_message_tests.rs

@@ -29,3 +29,35 @@ fn ignores_regular_user_text() {
         text: "hello".to_string(),
     }));
 }
+
+#[test]
+fn classifies_memory_excluded_fragments() {
+    let cases = [
+        (
+            "# AGENTS.md instructions for /tmp\n\n<INSTRUCTIONS>\nbody\n</INSTRUCTIONS>",
+            true,
+        ),
+        (
+            "<skill>\n<name>demo</name>\n<path>skills/demo/SKILL.md</path>\nbody\n</skill>",
+            true,
+        ),
+        (
+            "<environment_context>\n<cwd>/tmp</cwd>\n</environment_context>",
+            false,
+        ),
+        (
+            "<subagent_notification>{\"agent_id\":\"a\",\"status\":\"completed\"}</subagent_notification>",
+            false,
+        ),
+    ];
+
+    for (text, expected) in cases {
+        assert_eq!(
+            is_memory_excluded_contextual_user_fragment(&ContentItem::InputText {
+                text: text.to_string(),
+            }),
+            expected,
+            "{text}",
+        );
+    }
+}

codex-rs/core/src/mcp_connection_manager.rs

@@ -1231,12 +1231,11 @@ fn normalize_codex_apps_tool_name(
         return tool_name.to_string();
     }
 
-    let tool_name = sanitize_name(tool_name).replace('-', "_");
+    let tool_name = sanitize_name(tool_name);
 
     if let Some(connector_name) = connector_name
         .map(str::trim)
         .map(sanitize_name)
-        .map(|name| name.replace('-', "_"))
         .filter(|name| !name.is_empty())
         && let Some(stripped) = tool_name.strip_prefix(&connector_name)
         && !stripped.is_empty()
@@ -1247,7 +1246,6 @@ fn normalize_codex_apps_tool_name(
     if let Some(connector_id) = connector_id
         .map(str::trim)
         .map(sanitize_name)
-        .map(|name| name.replace('-', "_"))
         .filter(|name| !name.is_empty())
         && let Some(stripped) = tool_name.strip_prefix(&connector_id)
         && !stripped.is_empty()

codex-rs/core/src/memories/phase1.rs

@@ -4,6 +4,7 @@ use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::config::Config;
 use crate::config::types::MemoriesConfig;
+use crate::contextual_user_message::is_memory_excluded_contextual_user_fragment;
 use crate::error::CodexErr;
 use crate::memories::metrics;
 use crate::memories::phase_one;
@@ -463,16 +464,14 @@ mod job {
     }
 
     /// Serializes filtered stage-1 memory items for prompt inclusion.
-    fn serialize_filtered_rollout_response_items(
+    pub(super) fn serialize_filtered_rollout_response_items(
         items: &[RolloutItem],
     ) -> crate::error::Result<String> {
         let filtered = items
             .iter()
             .filter_map(|item| {
-                if let RolloutItem::ResponseItem(item) = item
-                    && should_persist_response_item_for_memories(item)
-                {
-                    Some(item.clone())
+                if let RolloutItem::ResponseItem(item) = item {
+                    sanitize_response_item_for_memories(item)
                 } else {
                     None
                 }
@@ -482,6 +481,44 @@ mod job {
             CodexErr::InvalidRequest(format!("failed to serialize rollout memory: {err}"))
         })
     }
+
+    fn sanitize_response_item_for_memories(item: &ResponseItem) -> Option<ResponseItem> {
+        let ResponseItem::Message {
+            id,
+            role,
+            content,
+            end_turn,
+            phase,
+        } = item
+        else {
+            return should_persist_response_item_for_memories(item).then(|| item.clone());
+        };
+
+        if role == "developer" {
+            return None;
+        }
+
+        if role != "user" {
+            return Some(item.clone());
+        }
+
+        let content = content
+            .iter()
+            .filter(|content_item| !is_memory_excluded_contextual_user_fragment(content_item))
+            .cloned()
+            .collect::<Vec<_>>();
+        if content.is_empty() {
+            return None;
+        }
+
+        Some(ResponseItem::Message {
+            id: id.clone(),
+            role: role.clone(),
+            content,
+            end_turn: *end_turn,
+            phase: phase.clone(),
+        })
+    }
 }
 
 fn aggregate_stats(outcomes: Vec<JobResult>) -> Stats {

codex-rs/core/src/memories/phase1_tests.rs

@@ -1,9 +1,77 @@
 use super::JobOutcome;
 use super::JobResult;
 use super::aggregate_stats;
+use super::job::serialize_filtered_rollout_response_items;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::TokenUsage;
 use pretty_assertions::assert_eq;
 
+#[test]
+fn serializes_memory_rollout_with_agents_removed_but_environment_kept() {
+    let mixed_contextual_message = ResponseItem::Message {
+        id: None,
+        role: "user".to_string(),
+        content: vec![
+            ContentItem::InputText {
+                text: "# AGENTS.md instructions for /tmp\n\n<INSTRUCTIONS>\nbody\n</INSTRUCTIONS>"
+                    .to_string(),
+            },
+            ContentItem::InputText {
+                text: "<environment_context>\n<cwd>/tmp</cwd>\n</environment_context>".to_string(),
+            },
+        ],
+        end_turn: None,
+        phase: None,
+    };
+    let skill_message = ResponseItem::Message {
+        id: None,
+        role: "user".to_string(),
+        content: vec![ContentItem::InputText {
+            text: "<skill>\n<name>demo</name>\n<path>skills/demo/SKILL.md</path>\nbody\n</skill>"
+                .to_string(),
+        }],
+        end_turn: None,
+        phase: None,
+    };
+    let subagent_message = ResponseItem::Message {
+        id: None,
+        role: "user".to_string(),
+        content: vec![ContentItem::InputText {
+            text: "<subagent_notification>{\"agent_id\":\"a\",\"status\":\"completed\"}</subagent_notification>"
+                .to_string(),
+        }],
+        end_turn: None,
+        phase: None,
+    };
+
+    let serialized = serialize_filtered_rollout_response_items(&[
+        RolloutItem::ResponseItem(mixed_contextual_message),
+        RolloutItem::ResponseItem(skill_message),
+        RolloutItem::ResponseItem(subagent_message.clone()),
+    ])
+    .expect("serialize");
+    let parsed: Vec<ResponseItem> = serde_json::from_str(&serialized).expect("parse");
+
+    assert_eq!(
+        parsed,
+        vec![
+            ResponseItem::Message {
+                id: None,
+                role: "user".to_string(),
+                content: vec![ContentItem::InputText {
+                    text: "<environment_context>\n<cwd>/tmp</cwd>\n</environment_context>"
+                        .to_string(),
+                }],
+                end_turn: None,
+                phase: None,
+            },
+            subagent_message,
+        ]
+    );
+}
+
 #[test]
 fn count_outcomes_sums_token_usage_across_all_jobs() {
     let counts = aggregate_stats(vec![

codex-rs/exec/src/lib.rs

@@ -36,10 +36,12 @@ use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadUnsubscribeParams;
 use codex_app_server_protocol::ThreadUnsubscribeResponse;
+use codex_app_server_protocol::TurnCompletedNotification;
 use codex_app_server_protocol::TurnInterruptParams;
 use codex_app_server_protocol::TurnInterruptResponse;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
+use codex_app_server_protocol::TurnStatus;
 use codex_arg0::Arg0DispatchPaths;
 use codex_cloud_requirements::cloud_requirements_loader;
 use codex_core::AuthManager;
@@ -70,6 +72,9 @@ use codex_protocol::protocol::ReviewRequest;
 use codex_protocol::protocol::ReviewTarget;
 use codex_protocol::protocol::SessionConfiguredEvent;
 use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::TurnAbortReason;
+use codex_protocol::protocol::TurnAbortedEvent;
+use codex_protocol::protocol::TurnCompleteEvent;
 use codex_protocol::user_input::UserInput;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_oss::ensure_oss_provider_ready;
@@ -782,6 +787,24 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
                 .await;
             }
             InProcessServerEvent::ServerNotification(notification) => {
+                if let Some((event, terminal_error)) = decode_turn_completed_notification_for_exec(
+                    &notification,
+                    primary_thread_id_for_requests.as_str(),
+                    task_id.as_str(),
+                ) {
+                    error_seen |= terminal_error;
+                    if handle_exec_status(
+                        event_processor.process_event(event),
+                        &client,
+                        &mut request_ids,
+                        &primary_thread_id_for_requests,
+                    )
+                    .await
+                    {
+                        break;
+                    }
+                    continue;
+                }
                 if let ServerNotification::Error(payload) = &notification
                     && payload.thread_id == primary_thread_id_for_requests
                     && payload.turn_id == task_id
@@ -850,25 +873,15 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
                     _ => {}
                 }
 
-                match event_processor.process_event(event) {
-                    CodexStatus::Running => {}
-                    CodexStatus::InitiateShutdown => {
-                        if let Err(err) = request_shutdown(
-                            &client,
-                            &mut request_ids,
-                            &primary_thread_id_for_requests,
-                        )
-                        .await
-                        {
-                            warn!("thread/unsubscribe failed during shutdown: {err}");
-                        }
-                        break;
-                    }
-                    CodexStatus::Shutdown => {
-                        // `ShutdownComplete` does not identify which attached
-                        // thread emitted it, so subagent shutdowns must not end
-                        // the primary exec loop early.
-                    }
+                if handle_exec_status(
+                    event_processor.process_event(event),
+                    &client,
+                    &mut request_ids,
+                    &primary_thread_id_for_requests,
+                )
+                .await
+                {
+                    break;
                 }
             }
             InProcessServerEvent::Lagged { skipped } => {
@@ -1131,6 +1144,80 @@ fn canceled_mcp_server_elicitation_response() -> Result<Value, String> {
     .map_err(|err| format!("failed to encode mcp elicitation response: {err}"))
 }
 
+async fn handle_exec_status(
+    status: CodexStatus,
+    client: &InProcessAppServerClient,
+    request_ids: &mut RequestIdSequencer,
+    primary_thread_id_for_requests: &str,
+) -> bool {
+    match status {
+        CodexStatus::Running => false,
+        CodexStatus::InitiateShutdown => {
+            if let Err(err) =
+                request_shutdown(client, request_ids, primary_thread_id_for_requests).await
+            {
+                warn!("thread/unsubscribe failed during shutdown: {err}");
+            }
+            true
+        }
+        CodexStatus::Shutdown => {
+            // `ShutdownComplete` does not identify which attached
+            // thread emitted it, so subagent shutdowns must not end
+            // the primary exec loop early.
+            false
+        }
+    }
+}
+
+fn decode_turn_completed_notification_for_exec(
+    notification: &ServerNotification,
+    primary_thread_id_for_requests: &str,
+    task_id: &str,
+) -> Option<(Event, bool)> {
+    let ServerNotification::TurnCompleted(TurnCompletedNotification { thread_id, turn }) =
+        notification
+    else {
+        return None;
+    };
+    if thread_id != primary_thread_id_for_requests || turn.id != task_id {
+        return None;
+    }
+
+    match turn.status {
+        TurnStatus::Completed => Some((
+            Event {
+                id: String::new(),
+                msg: EventMsg::TurnComplete(TurnCompleteEvent {
+                    turn_id: turn.id.clone(),
+                    last_agent_message: None,
+                }),
+            },
+            false,
+        )),
+        TurnStatus::Failed => Some((
+            Event {
+                id: String::new(),
+                msg: EventMsg::TurnComplete(TurnCompleteEvent {
+                    turn_id: turn.id.clone(),
+                    last_agent_message: None,
+                }),
+            },
+            true,
+        )),
+        TurnStatus::Interrupted => Some((
+            Event {
+                id: String::new(),
+                msg: EventMsg::TurnAborted(TurnAbortedEvent {
+                    turn_id: Some(turn.id.clone()),
+                    reason: TurnAbortReason::Interrupted,
+                }),
+            },
+            false,
+        )),
+        TurnStatus::InProgress => None,
+    }
+}
+
 async fn request_shutdown(
     client: &InProcessAppServerClient,
     request_ids: &mut RequestIdSequencer,
@@ -1922,4 +2009,118 @@ mod tests {
             ApprovalsReviewer::GuardianSubagent
         );
     }
+    #[test]
+    fn decode_turn_completed_notification_ignores_other_threads_and_turns() {
+        let thread_mismatch = ServerNotification::TurnCompleted(TurnCompletedNotification {
+            thread_id: "thread-a".to_string(),
+            turn: codex_app_server_protocol::Turn {
+                id: "turn-a".to_string(),
+                items: Vec::new(),
+                status: TurnStatus::Completed,
+                error: None,
+            },
+        });
+        assert!(
+            decode_turn_completed_notification_for_exec(&thread_mismatch, "thread-b", "turn-a")
+                .is_none()
+        );
+
+        let turn_mismatch = ServerNotification::TurnCompleted(TurnCompletedNotification {
+            thread_id: "thread-a".to_string(),
+            turn: codex_app_server_protocol::Turn {
+                id: "turn-a".to_string(),
+                items: Vec::new(),
+                status: TurnStatus::Completed,
+                error: None,
+            },
+        });
+        assert!(
+            decode_turn_completed_notification_for_exec(&turn_mismatch, "thread-a", "turn-b")
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn decode_turn_completed_notification_maps_completed_and_failed_turns() {
+        let completed_notification = ServerNotification::TurnCompleted(TurnCompletedNotification {
+            thread_id: "thread-a".to_string(),
+            turn: codex_app_server_protocol::Turn {
+                id: "turn-a".to_string(),
+                items: Vec::new(),
+                status: TurnStatus::Completed,
+                error: None,
+            },
+        });
+        let Some((completed, completed_error)) = decode_turn_completed_notification_for_exec(
+            &completed_notification,
+            "thread-a",
+            "turn-a",
+        ) else {
+            panic!("completed turn should decode");
+        };
+        assert!(!completed_error);
+        match completed.msg {
+            EventMsg::TurnComplete(event) => {
+                assert_eq!(event.turn_id, "turn-a");
+                assert_eq!(event.last_agent_message, None);
+            }
+            other => panic!("unexpected event: {other:?}"),
+        }
+
+        let failed_notification = ServerNotification::TurnCompleted(TurnCompletedNotification {
+            thread_id: "thread-a".to_string(),
+            turn: codex_app_server_protocol::Turn {
+                id: "turn-a".to_string(),
+                items: Vec::new(),
+                status: TurnStatus::Failed,
+                error: Some(codex_app_server_protocol::TurnError {
+                    message: "synthetic".to_string(),
+                    codex_error_info: None,
+                    additional_details: None,
+                }),
+            },
+        });
+        let Some((failed, failed_error)) =
+            decode_turn_completed_notification_for_exec(&failed_notification, "thread-a", "turn-a")
+        else {
+            panic!("failed turn should decode");
+        };
+        assert!(failed_error);
+        match failed.msg {
+            EventMsg::TurnComplete(event) => {
+                assert_eq!(event.turn_id, "turn-a");
+                assert_eq!(event.last_agent_message, None);
+            }
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn decode_turn_completed_notification_maps_interrupted_turns() {
+        let interrupted_notification =
+            ServerNotification::TurnCompleted(TurnCompletedNotification {
+                thread_id: "thread-a".to_string(),
+                turn: codex_app_server_protocol::Turn {
+                    id: "turn-a".to_string(),
+                    items: Vec::new(),
+                    status: TurnStatus::Interrupted,
+                    error: None,
+                },
+            });
+        let Some((event, terminal_error)) = decode_turn_completed_notification_for_exec(
+            &interrupted_notification,
+            "thread-a",
+            "turn-a",
+        ) else {
+            panic!("interrupted turn should decode");
+        };
+        assert!(!terminal_error);
+        match event.msg {
+            EventMsg::TurnAborted(event) => {
+                assert_eq!(event.turn_id.as_deref(), Some("turn-a"));
+                assert_eq!(event.reason, TurnAbortReason::Interrupted);
+            }
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
 }

codex-rs/exec/tests/suite/agent_jobs.rs

@@ -0,0 +1,237 @@
+#![cfg(not(target_os = "windows"))]
+#![allow(clippy::expect_used, clippy::unwrap_used)]
+
+use anyhow::Result;
+use core_test_support::test_codex_exec::test_codex_exec;
+use serde_json::Value;
+use serde_json::json;
+use std::fs;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+use wiremock::Mock;
+use wiremock::Respond;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path_regex;
+
+struct AgentJobsResponder {
+    spawn_args_json: String,
+    seen_main: AtomicBool,
+    call_counter: AtomicUsize,
+}
+
+impl AgentJobsResponder {
+    fn new(spawn_args_json: String) -> Self {
+        Self {
+            spawn_args_json,
+            seen_main: AtomicBool::new(false),
+            call_counter: AtomicUsize::new(0),
+        }
+    }
+}
+
+impl Respond for AgentJobsResponder {
+    fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
+        let body_bytes = decode_body_bytes(request);
+        let body: Value = serde_json::from_slice(&body_bytes).unwrap_or(Value::Null);
+
+        if has_function_call_output(&body) {
+            return sse_response(sse(vec![
+                ev_response_created("resp-tool"),
+                ev_completed("resp-tool"),
+            ]));
+        }
+
+        if let Some((job_id, item_id)) = extract_job_and_item(&body) {
+            let call_id = format!(
+                "call-worker-{}",
+                self.call_counter.fetch_add(1, Ordering::SeqCst)
+            );
+            let args = json!({
+                "job_id": job_id,
+                "item_id": item_id,
+                "result": { "item_id": item_id }
+            });
+            let args_json = serde_json::to_string(&args).unwrap_or_else(|err| {
+                panic!("worker args serialize: {err}");
+            });
+            return sse_response(sse(vec![
+                ev_response_created("resp-worker"),
+                ev_function_call(&call_id, "report_agent_job_result", &args_json),
+                ev_completed("resp-worker"),
+            ]));
+        }
+
+        if !self.seen_main.swap(true, Ordering::SeqCst) {
+            return sse_response(sse(vec![
+                ev_response_created("resp-main"),
+                ev_function_call("call-spawn", "spawn_agents_on_csv", &self.spawn_args_json),
+                ev_completed("resp-main"),
+            ]));
+        }
+
+        sse_response(sse(vec![
+            ev_response_created("resp-default"),
+            ev_completed("resp-default"),
+        ]))
+    }
+}
+
+fn decode_body_bytes(request: &wiremock::Request) -> Vec<u8> {
+    request.body.clone()
+}
+
+fn has_function_call_output(body: &Value) -> bool {
+    body.get("input")
+        .and_then(Value::as_array)
+        .is_some_and(|items| {
+            items.iter().any(|item| {
+                item.get("type").and_then(Value::as_str) == Some("function_call_output")
+            })
+        })
+}
+
+fn extract_job_and_item(body: &Value) -> Option<(String, String)> {
+    let texts = message_input_texts(body);
+    let mut combined = texts.join("\n");
+    if let Some(instructions) = body.get("instructions").and_then(Value::as_str) {
+        combined.push('\n');
+        combined.push_str(instructions);
+    }
+    if !combined.contains("You are processing one item for a generic agent job.") {
+        return None;
+    }
+
+    let mut job_id = None;
+    let mut item_id = None;
+    for line in combined.lines() {
+        if let Some(value) = line.strip_prefix("Job ID: ") {
+            job_id = Some(value.trim().to_string());
+        }
+        if let Some(value) = line.strip_prefix("Item ID: ") {
+            item_id = Some(value.trim().to_string());
+        }
+    }
+
+    Some((job_id?, item_id?))
+}
+
+fn message_input_texts(body: &Value) -> Vec<String> {
+    let Some(items) = body.get("input").and_then(Value::as_array) else {
+        return Vec::new();
+    };
+    items
+        .iter()
+        .filter(|item| item.get("type").and_then(Value::as_str) == Some("message"))
+        .filter_map(|item| item.get("content").and_then(Value::as_array))
+        .flatten()
+        .filter(|span| span.get("type").and_then(Value::as_str) == Some("input_text"))
+        .filter_map(|span| span.get("text").and_then(Value::as_str))
+        .map(str::to_string)
+        .collect()
+}
+
+fn sse(events: Vec<serde_json::Value>) -> String {
+    let mut body = String::new();
+    for event in events {
+        body.push_str("data: ");
+        body.push_str(&event.to_string());
+        body.push_str("\n\n");
+    }
+    body.push_str("data: [DONE]\n\n");
+    body
+}
+
+fn sse_response(body: String) -> ResponseTemplate {
+    ResponseTemplate::new(200)
+        .insert_header("content-type", "text/event-stream")
+        .set_body_string(body)
+}
+
+fn ev_response_created(response_id: &str) -> serde_json::Value {
+    json!({
+        "type": "response.created",
+        "response": {
+            "id": response_id,
+            "model": "gpt-5",
+            "output": []
+        }
+    })
+}
+
+fn ev_function_call(call_id: &str, name: &str, arguments: &str) -> serde_json::Value {
+    json!({
+        "type": "response.output_item.done",
+        "output_index": 0,
+        "item": {
+            "type": "function_call",
+            "id": format!("item-{call_id}"),
+            "call_id": call_id,
+            "name": name,
+            "arguments": arguments,
+            "status": "completed"
+        }
+    })
+}
+
+fn ev_completed(response_id: &str) -> serde_json::Value {
+    json!({
+        "type": "response.completed",
+        "response": {
+            "id": response_id,
+            "usage": {
+                "input_tokens": 1,
+                "input_tokens_details": {"cached_tokens": 0},
+                "output_tokens": 1,
+                "output_tokens_details": {"reasoning_tokens": 0},
+                "total_tokens": 2
+            }
+        }
+    })
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn exec_spawn_agents_on_csv_exits_after_mock_job_completion() -> Result<()> {
+    let test = test_codex_exec();
+    let server = wiremock::MockServer::start().await;
+
+    let input_path = test.cwd_path().join("agent_jobs_input.csv");
+    let output_path = test.cwd_path().join("agent_jobs_output.csv");
+    let mut csv = String::from("name\n");
+    for index in 1..=100 {
+        csv.push_str(&format!("cat_{index}\n"));
+    }
+    fs::write(&input_path, csv)?;
+
+    let args = json!({
+        "csv_path": input_path.display().to_string(),
+        "instruction": "Write a playful 2-line poem about the cat named {name}. Return a JSON object with keys name and poem. Call report_agent_job_result exactly once and then stop.",
+        "output_csv_path": output_path.display().to_string(),
+        "max_concurrency": 64,
+    });
+    let args_json = serde_json::to_string(&args)?;
+
+    Mock::given(method("POST"))
+        .and(path_regex(".*/responses$"))
+        .respond_with(AgentJobsResponder::new(args_json))
+        .mount(&server)
+        .await;
+
+    let mut cmd = test.cmd_with_server(&server);
+    cmd.timeout(Duration::from_secs(60));
+    cmd.arg("-c")
+        .arg("features.enable_fanout=true")
+        .arg("-c")
+        .arg("agents.max_threads=64")
+        .arg("--skip-git-repo-check")
+        .arg("Use spawn_agents_on_csv on the provided CSV and do not do work yourself.")
+        .assert()
+        .success();
+
+    let output = fs::read_to_string(&output_path)?;
+    assert_eq!(output.lines().count(), 101);
+
+    Ok(())
+}

codex-rs/exec/tests/suite/mod.rs

@@ -1,5 +1,6 @@
 // Aggregates all former standalone integration tests as modules.
 mod add_dir;
+mod agent_jobs;
 mod apply_patch;
 mod auth_env;
 mod ephemeral;

codex-rs/utils/pty/src/lib.rs

@@ -29,3 +29,5 @@ pub type SpawnedPty = SpawnedProcess;
 pub use pty::conpty_supported;
 /// Spawn a process attached to a PTY for interactive use.
 pub use pty::spawn_process as spawn_pty_process;
+#[cfg(windows)]
+pub use win::conpty::RawConPty;

codex-rs/utils/pty/src/win/conpty.rs

@@ -29,6 +29,9 @@ use portable_pty::PtyPair;
 use portable_pty::PtySize;
 use portable_pty::PtySystem;
 use portable_pty::SlavePty;
+use std::mem::ManuallyDrop;
+use std::os::windows::io::AsRawHandle;
+use std::os::windows::io::RawHandle;
 use std::sync::Arc;
 use std::sync::Mutex;
 use winapi::um::wincon::COORD;
@@ -36,25 +39,68 @@ use winapi::um::wincon::COORD;
 #[derive(Default)]
 pub struct ConPtySystem {}
 
+fn create_conpty_handles(
+    size: PtySize,
+) -> anyhow::Result<(PsuedoCon, FileDescriptor, FileDescriptor)> {
+    let stdin = Pipe::new()?;
+    let stdout = Pipe::new()?;
+
+    let con = PsuedoCon::new(
+        COORD {
+            X: size.cols as i16,
+            Y: size.rows as i16,
+        },
+        stdin.read,
+        stdout.write,
+    )?;
+
+    Ok((con, stdin.write, stdout.read))
+}
+
+pub struct RawConPty {
+    con: PsuedoCon,
+    input_write: FileDescriptor,
+    output_read: FileDescriptor,
+}
+
+impl RawConPty {
+    pub fn new(cols: i16, rows: i16) -> anyhow::Result<Self> {
+        let (con, input_write, output_read) = create_conpty_handles(PtySize {
+            rows: rows as u16,
+            cols: cols as u16,
+            pixel_width: 0,
+            pixel_height: 0,
+        })?;
+        Ok(Self {
+            con,
+            input_write,
+            output_read,
+        })
+    }
+
+    pub fn pseudoconsole_handle(&self) -> RawHandle {
+        self.con.raw_handle()
+    }
+
+    pub fn into_raw_handles(self) -> (RawHandle, RawHandle, RawHandle) {
+        let me = ManuallyDrop::new(self);
+        (
+            me.con.raw_handle(),
+            me.input_write.as_raw_handle(),
+            me.output_read.as_raw_handle(),
+        )
+    }
+}
+
 impl PtySystem for ConPtySystem {
     fn openpty(&self, size: PtySize) -> anyhow::Result<PtyPair> {
-        let stdin = Pipe::new()?;
-        let stdout = Pipe::new()?;
-
-        let con = PsuedoCon::new(
-            COORD {
-                X: size.cols as i16,
-                Y: size.rows as i16,
-            },
-            stdin.read,
-            stdout.write,
-        )?;
+        let (con, writable, readable) = create_conpty_handles(size)?;
 
         let master = ConPtyMasterPty {
             inner: Arc::new(Mutex::new(Inner {
                 con,
-                readable: stdout.read,
-                writable: Some(stdin.write),
+                readable,
+                writable: Some(writable),
                 size,
             })),
         };

codex-rs/utils/pty/src/win/psuedocon.rs

@@ -130,6 +130,10 @@ impl Drop for PsuedoCon {
 }
 
 impl PsuedoCon {
+    pub fn raw_handle(&self) -> HPCON {
+        self.con
+    }
+
     pub fn new(size: COORD, input: FileDescriptor, output: FileDescriptor) -> Result<Self, Error> {
         let mut con: HPCON = INVALID_HANDLE_VALUE;
         let result = unsafe {

codex-rs/windows-sandbox-rs/Cargo.toml

@@ -24,12 +24,14 @@ chrono = { version = "0.4.42", default-features = false, features = [
     "clock",
     "std",
 ] }
+codex-utils-pty = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-string = { workspace = true }
 dunce = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tempfile = "3"
+tokio = { workspace = true, features = ["sync", "rt"] }
 windows = { version = "0.58", features = [
     "Win32_Foundation",
     "Win32_NetworkManagement_WindowsFirewall",
@@ -86,3 +88,6 @@ pretty_assertions = { workspace = true }
 
 [build-dependencies]
 winres = "0.1"
+
+[package.metadata.cargo-shear]
+ignored = ["codex-utils-pty", "tokio"]

codex-rs/windows-sandbox-rs/src/bin/command_runner.rs

@@ -1,4 +1,4 @@
-#[path = "../command_runner_win.rs"]
+#[path = "../elevated/command_runner_win.rs"]
 mod win;
 
 #[cfg(target_os = "windows")]

codex-rs/windows-sandbox-rs/src/command_runner_win.rs

@@ -1,325 +0,0 @@
-#![cfg(target_os = "windows")]
-
-use anyhow::Context;
-use anyhow::Result;
-use codex_windows_sandbox::allow_null_device;
-use codex_windows_sandbox::convert_string_sid_to_sid;
-use codex_windows_sandbox::create_process_as_user;
-use codex_windows_sandbox::create_readonly_token_with_caps_from;
-use codex_windows_sandbox::create_workspace_write_token_with_caps_from;
-use codex_windows_sandbox::get_current_token_for_restriction;
-use codex_windows_sandbox::hide_current_user_profile_dir;
-use codex_windows_sandbox::log_note;
-use codex_windows_sandbox::parse_policy;
-use codex_windows_sandbox::to_wide;
-use codex_windows_sandbox::SandboxPolicy;
-use serde::Deserialize;
-use std::collections::HashMap;
-use std::ffi::c_void;
-use std::path::Path;
-use std::path::PathBuf;
-use windows_sys::Win32::Foundation::CloseHandle;
-use windows_sys::Win32::Foundation::GetLastError;
-use windows_sys::Win32::Foundation::LocalFree;
-use windows_sys::Win32::Foundation::HANDLE;
-use windows_sys::Win32::Foundation::HLOCAL;
-use windows_sys::Win32::Storage::FileSystem::CreateFileW;
-use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_READ;
-use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_WRITE;
-use windows_sys::Win32::Storage::FileSystem::OPEN_EXISTING;
-use windows_sys::Win32::System::Diagnostics::Debug::SetErrorMode;
-use windows_sys::Win32::System::JobObjects::AssignProcessToJobObject;
-use windows_sys::Win32::System::JobObjects::CreateJobObjectW;
-use windows_sys::Win32::System::JobObjects::JobObjectExtendedLimitInformation;
-use windows_sys::Win32::System::JobObjects::SetInformationJobObject;
-use windows_sys::Win32::System::JobObjects::JOBOBJECT_EXTENDED_LIMIT_INFORMATION;
-use windows_sys::Win32::System::JobObjects::JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
-use windows_sys::Win32::System::Threading::TerminateProcess;
-use windows_sys::Win32::System::Threading::WaitForSingleObject;
-use windows_sys::Win32::System::Threading::INFINITE;
-
-#[path = "cwd_junction.rs"]
-mod cwd_junction;
-
-#[allow(dead_code)]
-mod read_acl_mutex;
-
-#[derive(Debug, Deserialize)]
-struct RunnerRequest {
-    policy_json_or_preset: String,
-    // Writable location for logs (sandbox user's .codex).
-    codex_home: PathBuf,
-    // Real user's CODEX_HOME for shared data (caps, config).
-    real_codex_home: PathBuf,
-    cap_sids: Vec<String>,
-    command: Vec<String>,
-    cwd: PathBuf,
-    env_map: HashMap<String, String>,
-    timeout_ms: Option<u64>,
-    use_private_desktop: bool,
-    stdin_pipe: String,
-    stdout_pipe: String,
-    stderr_pipe: String,
-}
-
-const WAIT_TIMEOUT: u32 = 0x0000_0102;
-
-unsafe fn create_job_kill_on_close() -> Result<HANDLE> {
-    let h = CreateJobObjectW(std::ptr::null_mut(), std::ptr::null());
-    if h == 0 {
-        return Err(anyhow::anyhow!("CreateJobObjectW failed"));
-    }
-    let mut limits: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed();
-    limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
-    let ok = SetInformationJobObject(
-        h,
-        JobObjectExtendedLimitInformation,
-        &mut limits as *mut _ as *mut _,
-        std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
-    );
-    if ok == 0 {
-        return Err(anyhow::anyhow!("SetInformationJobObject failed"));
-    }
-    Ok(h)
-}
-
-fn read_request_file(req_path: &Path) -> Result<String> {
-    let content = std::fs::read_to_string(req_path)
-        .with_context(|| format!("read request file {}", req_path.display()));
-    let _ = std::fs::remove_file(req_path);
-    content
-}
-
-pub fn main() -> Result<()> {
-    let mut input = String::new();
-    let mut args = std::env::args().skip(1);
-    if let Some(first) = args.next() {
-        if let Some(rest) = first.strip_prefix("--request-file=") {
-            let req_path = PathBuf::from(rest);
-            input = read_request_file(&req_path)?;
-        }
-    }
-    if input.is_empty() {
-        anyhow::bail!("runner: no request-file provided");
-    }
-    let req: RunnerRequest = serde_json::from_str(&input).context("parse runner request json")?;
-    let log_dir = Some(req.codex_home.as_path());
-    hide_current_user_profile_dir(req.codex_home.as_path());
-    // Suppress Windows error UI from sandboxed child crashes so callers only observe exit codes.
-    let _ = unsafe { SetErrorMode(0x0001 | 0x0002) }; // SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX
-    log_note(
-        &format!(
-            "runner start cwd={} cmd={:?} real_codex_home={}",
-            req.cwd.display(),
-            req.command,
-            req.real_codex_home.display()
-        ),
-        Some(&req.codex_home),
-    );
-
-    let policy = parse_policy(&req.policy_json_or_preset).context("parse policy_json_or_preset")?;
-    if !policy.has_full_disk_read_access() {
-        anyhow::bail!(
-            "Restricted read-only access is not yet supported by the Windows sandbox backend"
-        );
-    }
-    let mut cap_psids: Vec<*mut c_void> = Vec::new();
-    for sid in &req.cap_sids {
-        let Some(psid) = (unsafe { convert_string_sid_to_sid(sid) }) else {
-            anyhow::bail!("ConvertStringSidToSidW failed for capability SID");
-        };
-        cap_psids.push(psid);
-    }
-    if cap_psids.is_empty() {
-        anyhow::bail!("runner: empty capability SID list");
-    }
-
-    // Create restricted token from current process token.
-    let base = unsafe { get_current_token_for_restriction()? };
-    let token_res: Result<HANDLE> = unsafe {
-        match &policy {
-            SandboxPolicy::ReadOnly { .. } => {
-                create_readonly_token_with_caps_from(base, &cap_psids)
-            }
-            SandboxPolicy::WorkspaceWrite { .. } => {
-                create_workspace_write_token_with_caps_from(base, &cap_psids)
-            }
-            SandboxPolicy::DangerFullAccess | SandboxPolicy::ExternalSandbox { .. } => {
-                unreachable!()
-            }
-        }
-    };
-    let h_token = token_res?;
-    unsafe {
-        CloseHandle(base);
-    }
-    unsafe {
-        for psid in &cap_psids {
-            allow_null_device(*psid);
-        }
-        for psid in cap_psids {
-            if !psid.is_null() {
-                LocalFree(psid as HLOCAL);
-            }
-        }
-    }
-
-    // Open named pipes for stdio.
-    let open_pipe = |name: &str, access: u32| -> Result<HANDLE> {
-        let path = to_wide(name);
-        let handle = unsafe {
-            CreateFileW(
-                path.as_ptr(),
-                access,
-                0,
-                std::ptr::null_mut(),
-                OPEN_EXISTING,
-                0,
-                0,
-            )
-        };
-        if handle == windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE {
-            let err = unsafe { GetLastError() };
-            log_note(
-                &format!("CreateFileW failed for pipe {name}: {err}"),
-                Some(&req.codex_home),
-            );
-            return Err(anyhow::anyhow!("CreateFileW failed for pipe {name}: {err}"));
-        }
-        Ok(handle)
-    };
-    let h_stdin = open_pipe(&req.stdin_pipe, FILE_GENERIC_READ)?;
-    let h_stdout = open_pipe(&req.stdout_pipe, FILE_GENERIC_WRITE)?;
-    let h_stderr = open_pipe(&req.stderr_pipe, FILE_GENERIC_WRITE)?;
-    let stdio = Some((h_stdin, h_stdout, h_stderr));
-
-    // While the read-ACL helper is running, PowerShell can fail to start in the requested CWD due
-    // to unreadable ancestors. Use a junction CWD for that window; once the helper finishes, go
-    // back to using the real requested CWD (no probing, no extra state).
-    let use_junction = match read_acl_mutex::read_acl_mutex_exists() {
-        Ok(exists) => exists,
-        Err(err) => {
-            // Fail-safe: if we can't determine the state, assume the helper might be running and
-            // use the junction path to avoid CWD failures on unreadable ancestors.
-            log_note(
-                &format!("junction: read_acl_mutex_exists failed: {err}; assuming read ACL helper is running"),
-                log_dir,
-            );
-            true
-        }
-    };
-    if use_junction {
-        log_note(
-            "junction: read ACL helper running; using junction CWD",
-            log_dir,
-        );
-    }
-    let effective_cwd = if use_junction {
-        cwd_junction::create_cwd_junction(&req.cwd, log_dir).unwrap_or_else(|| req.cwd.clone())
-    } else {
-        req.cwd.clone()
-    };
-    log_note(
-        &format!(
-            "runner: effective cwd={} (requested {})",
-            effective_cwd.display(),
-            req.cwd.display()
-        ),
-        log_dir,
-    );
-
-    // Build command and env, spawn with CreateProcessAsUserW.
-    let spawn_result = unsafe {
-        create_process_as_user(
-            h_token,
-            &req.command,
-            &effective_cwd,
-            &req.env_map,
-            Some(&req.codex_home),
-            stdio,
-            req.use_private_desktop,
-        )
-    };
-    let created = match spawn_result {
-        Ok(v) => v,
-        Err(e) => {
-            log_note(&format!("runner: spawn failed: {e:?}"), log_dir);
-            unsafe {
-                CloseHandle(h_stdin);
-                CloseHandle(h_stdout);
-                CloseHandle(h_stderr);
-                CloseHandle(h_token);
-            }
-            return Err(e);
-        }
-    };
-    let proc_info = created.process_info;
-    let _desktop = created;
-
-    // Optional job kill on close.
-    let h_job = unsafe { create_job_kill_on_close().ok() };
-    if let Some(job) = h_job {
-        unsafe {
-            let _ = AssignProcessToJobObject(job, proc_info.hProcess);
-        }
-    }
-
-    // Wait for process.
-    let wait_res = unsafe {
-        WaitForSingleObject(
-            proc_info.hProcess,
-            req.timeout_ms.map(|ms| ms as u32).unwrap_or(INFINITE),
-        )
-    };
-    let timed_out = wait_res == WAIT_TIMEOUT;
-
-    let exit_code: i32;
-    unsafe {
-        if timed_out {
-            let _ = TerminateProcess(proc_info.hProcess, 1);
-            exit_code = 128 + 64;
-        } else {
-            let mut raw_exit: u32 = 1;
-            windows_sys::Win32::System::Threading::GetExitCodeProcess(
-                proc_info.hProcess,
-                &mut raw_exit,
-            );
-            exit_code = raw_exit as i32;
-        }
-        if proc_info.hThread != 0 {
-            CloseHandle(proc_info.hThread);
-        }
-        if proc_info.hProcess != 0 {
-            CloseHandle(proc_info.hProcess);
-        }
-        CloseHandle(h_stdin);
-        CloseHandle(h_stdout);
-        CloseHandle(h_stderr);
-        CloseHandle(h_token);
-        if let Some(job) = h_job {
-            CloseHandle(job);
-        }
-    }
-    if exit_code != 0 {
-        eprintln!("runner child exited with code {}", exit_code);
-    }
-    std::process::exit(exit_code);
-}
-
-#[cfg(test)]
-mod tests {
-    use super::read_request_file;
-    use pretty_assertions::assert_eq;
-    use std::fs;
-
-    #[test]
-    fn removes_request_file_after_read() {
-        let dir = tempfile::tempdir().expect("tempdir");
-        let req_path = dir.path().join("request.json");
-        fs::write(&req_path, "{\"ok\":true}").expect("write request");
-
-        let content = read_request_file(&req_path).expect("read request");
-        assert_eq!(content, "{\"ok\":true}");
-        assert!(!req_path.exists(), "request file should be removed");
-    }
-}

codex-rs/windows-sandbox-rs/src/conpty/mod.rs

@@ -0,0 +1,139 @@
+//! ConPTY helpers for spawning sandboxed processes with a PTY on Windows.
+//!
+//! This module encapsulates ConPTY creation and process spawn with the required
+//! `PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE` plumbing. It is shared by both the legacy
+//! restricted‑token path and the elevated runner path when unified_exec runs with
+//! `tty=true`. The helpers are not tied to the IPC layer and can be reused by other
+//! Windows sandbox flows that need a PTY.
+
+mod proc_thread_attr;
+
+use self::proc_thread_attr::ProcThreadAttributeList;
+use crate::winutil::format_last_error;
+use crate::winutil::quote_windows_arg;
+use crate::winutil::to_wide;
+use anyhow::Result;
+use codex_utils_pty::RawConPty;
+use std::collections::HashMap;
+use std::ffi::c_void;
+use std::path::Path;
+use windows_sys::Win32::Foundation::CloseHandle;
+use windows_sys::Win32::Foundation::GetLastError;
+use windows_sys::Win32::Foundation::HANDLE;
+use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
+use windows_sys::Win32::System::Console::ClosePseudoConsole;
+use windows_sys::Win32::System::Threading::CreateProcessAsUserW;
+use windows_sys::Win32::System::Threading::CREATE_UNICODE_ENVIRONMENT;
+use windows_sys::Win32::System::Threading::EXTENDED_STARTUPINFO_PRESENT;
+use windows_sys::Win32::System::Threading::PROCESS_INFORMATION;
+use windows_sys::Win32::System::Threading::STARTF_USESTDHANDLES;
+use windows_sys::Win32::System::Threading::STARTUPINFOEXW;
+
+use crate::process::make_env_block;
+
+/// Owns a ConPTY handle and its backing pipe handles.
+pub struct ConptyInstance {
+    pub hpc: HANDLE,
+    pub input_write: HANDLE,
+    pub output_read: HANDLE,
+}
+
+impl Drop for ConptyInstance {
+    fn drop(&mut self) {
+        unsafe {
+            if self.input_write != 0 && self.input_write != INVALID_HANDLE_VALUE {
+                CloseHandle(self.input_write);
+            }
+            if self.output_read != 0 && self.output_read != INVALID_HANDLE_VALUE {
+                CloseHandle(self.output_read);
+            }
+            if self.hpc != 0 && self.hpc != INVALID_HANDLE_VALUE {
+                ClosePseudoConsole(self.hpc);
+            }
+        }
+    }
+}
+
+impl ConptyInstance {
+    /// Consume the instance and return raw handles without closing them.
+    pub fn into_raw(self) -> (HANDLE, HANDLE, HANDLE) {
+        let me = std::mem::ManuallyDrop::new(self);
+        (me.hpc, me.input_write, me.output_read)
+    }
+}
+
+/// Create a ConPTY with backing pipes.
+///
+/// This is public so callers that need lower-level PTY setup can build on the same
+/// primitive, although the common entry point is `spawn_conpty_process_as_user`.
+pub fn create_conpty(cols: i16, rows: i16) -> Result<ConptyInstance> {
+    let raw = RawConPty::new(cols, rows)?;
+    let (hpc, input_write, output_read) = raw.into_raw_handles();
+
+    Ok(ConptyInstance {
+        hpc: hpc as HANDLE,
+        input_write: input_write as HANDLE,
+        output_read: output_read as HANDLE,
+    })
+}
+
+/// Spawn a process under `h_token` with ConPTY attached.
+///
+/// This is the main shared ConPTY entry point and is used by both the legacy/direct path
+/// and the elevated runner path whenever a PTY-backed sandboxed process is needed.
+pub fn spawn_conpty_process_as_user(
+    h_token: HANDLE,
+    argv: &[String],
+    cwd: &Path,
+    env_map: &HashMap<String, String>,
+) -> Result<(PROCESS_INFORMATION, ConptyInstance)> {
+    let cmdline_str = argv
+        .iter()
+        .map(|arg| quote_windows_arg(arg))
+        .collect::<Vec<_>>()
+        .join(" ");
+    let mut cmdline: Vec<u16> = to_wide(&cmdline_str);
+    let env_block = make_env_block(env_map);
+    let mut si: STARTUPINFOEXW = unsafe { std::mem::zeroed() };
+    si.StartupInfo.cb = std::mem::size_of::<STARTUPINFOEXW>() as u32;
+    si.StartupInfo.dwFlags = STARTF_USESTDHANDLES;
+    si.StartupInfo.hStdInput = INVALID_HANDLE_VALUE;
+    si.StartupInfo.hStdOutput = INVALID_HANDLE_VALUE;
+    si.StartupInfo.hStdError = INVALID_HANDLE_VALUE;
+    let desktop = to_wide("Winsta0\\Default");
+    si.StartupInfo.lpDesktop = desktop.as_ptr() as *mut u16;
+
+    let conpty = create_conpty(80, 24)?;
+    let mut attrs = ProcThreadAttributeList::new(1)?;
+    attrs.set_pseudoconsole(conpty.hpc)?;
+    si.lpAttributeList = attrs.as_mut_ptr();
+
+    let mut pi: PROCESS_INFORMATION = unsafe { std::mem::zeroed() };
+    let ok = unsafe {
+        CreateProcessAsUserW(
+            h_token,
+            std::ptr::null(),
+            cmdline.as_mut_ptr(),
+            std::ptr::null_mut(),
+            std::ptr::null_mut(),
+            0,
+            EXTENDED_STARTUPINFO_PRESENT | CREATE_UNICODE_ENVIRONMENT,
+            env_block.as_ptr() as *mut c_void,
+            to_wide(cwd).as_ptr(),
+            &si.StartupInfo,
+            &mut pi,
+        )
+    };
+    if ok == 0 {
+        let err = unsafe { GetLastError() } as i32;
+        return Err(anyhow::anyhow!(
+            "CreateProcessAsUserW failed: {} ({}) | cwd={} | cmd={} | env_u16_len={}",
+            err,
+            format_last_error(err),
+            cwd.display(),
+            cmdline_str,
+            env_block.len()
+        ));
+    }
+    Ok((pi, conpty))
+}

codex-rs/windows-sandbox-rs/src/conpty/proc_thread_attr.rs

@@ -0,0 +1,79 @@
+//! Low-level Windows thread attribute helpers used by ConPTY spawn.
+//!
+//! This module wraps the Win32 `PROC_THREAD_ATTRIBUTE_LIST` APIs so ConPTY handles can
+//! be attached to a child process. It is ConPTY‑specific and used in both legacy and
+//! elevated unified_exec paths when spawning a PTY‑backed process.
+
+use std::io;
+use windows_sys::Win32::Foundation::GetLastError;
+use windows_sys::Win32::System::Threading::DeleteProcThreadAttributeList;
+use windows_sys::Win32::System::Threading::InitializeProcThreadAttributeList;
+use windows_sys::Win32::System::Threading::UpdateProcThreadAttribute;
+use windows_sys::Win32::System::Threading::LPPROC_THREAD_ATTRIBUTE_LIST;
+
+const PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE: usize = 0x00020016;
+
+/// RAII wrapper for Windows PROC_THREAD_ATTRIBUTE_LIST.
+pub struct ProcThreadAttributeList {
+    buffer: Vec<u8>,
+}
+
+impl ProcThreadAttributeList {
+    /// Allocate and initialize a thread attribute list.
+    pub fn new(attr_count: u32) -> io::Result<Self> {
+        let mut size: usize = 0;
+        unsafe {
+            InitializeProcThreadAttributeList(std::ptr::null_mut(), attr_count, 0, &mut size);
+        }
+        if size == 0 {
+            return Err(io::Error::from_raw_os_error(unsafe {
+                GetLastError() as i32
+            }));
+        }
+        let mut buffer = vec![0u8; size];
+        let list = buffer.as_mut_ptr() as LPPROC_THREAD_ATTRIBUTE_LIST;
+        let ok = unsafe { InitializeProcThreadAttributeList(list, attr_count, 0, &mut size) };
+        if ok == 0 {
+            return Err(io::Error::from_raw_os_error(unsafe {
+                GetLastError() as i32
+            }));
+        }
+        Ok(Self { buffer })
+    }
+
+    /// Return a mutable pointer to the attribute list for Win32 APIs.
+    pub fn as_mut_ptr(&mut self) -> LPPROC_THREAD_ATTRIBUTE_LIST {
+        self.buffer.as_mut_ptr() as LPPROC_THREAD_ATTRIBUTE_LIST
+    }
+
+    /// Attach a ConPTY handle to the attribute list.
+    pub fn set_pseudoconsole(&mut self, hpc: isize) -> io::Result<()> {
+        let list = self.as_mut_ptr();
+        let mut hpc_value = hpc;
+        let ok = unsafe {
+            UpdateProcThreadAttribute(
+                list,
+                0,
+                PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE,
+                (&mut hpc_value as *mut isize).cast(),
+                std::mem::size_of::<isize>(),
+                std::ptr::null_mut(),
+                std::ptr::null_mut(),
+            )
+        };
+        if ok == 0 {
+            return Err(io::Error::from_raw_os_error(unsafe {
+                GetLastError() as i32
+            }));
+        }
+        Ok(())
+    }
+}
+
+impl Drop for ProcThreadAttributeList {
+    fn drop(&mut self) {
+        unsafe {
+            DeleteProcThreadAttributeList(self.as_mut_ptr());
+        }
+    }
+}

codex-rs/windows-sandbox-rs/src/elevated/command_runner_win.rs

@@ -0,0 +1,742 @@
+//! Windows command runner used by the **elevated** sandbox path.
+//!
+//! The CLI launches this binary under the sandbox user when Windows sandbox level is
+//! Elevated. It connects to the IPC pipes, reads the framed `SpawnRequest`, derives a
+//! restricted token from the sandbox user, and spawns the child process via ConPTY
+//! (`tty=true`) or pipes (`tty=false`). It then streams output frames back to the parent,
+//! accepts stdin/terminate frames, and emits a final exit frame. The legacy restricted‑token
+//! path spawns the child directly and does not use this runner.
+
+#![cfg(target_os = "windows")]
+
+use anyhow::Context;
+use anyhow::Result;
+use codex_windows_sandbox::allow_null_device;
+use codex_windows_sandbox::convert_string_sid_to_sid;
+use codex_windows_sandbox::create_process_as_user;
+use codex_windows_sandbox::create_readonly_token_with_caps_from;
+use codex_windows_sandbox::create_workspace_write_token_with_caps_from;
+use codex_windows_sandbox::get_current_token_for_restriction;
+use codex_windows_sandbox::hide_current_user_profile_dir;
+use codex_windows_sandbox::ipc_framed::decode_bytes;
+use codex_windows_sandbox::ipc_framed::encode_bytes;
+use codex_windows_sandbox::ipc_framed::read_frame;
+use codex_windows_sandbox::ipc_framed::write_frame;
+use codex_windows_sandbox::ipc_framed::ErrorPayload;
+use codex_windows_sandbox::ipc_framed::ExitPayload;
+use codex_windows_sandbox::ipc_framed::FramedMessage;
+use codex_windows_sandbox::ipc_framed::Message;
+use codex_windows_sandbox::ipc_framed::OutputPayload;
+use codex_windows_sandbox::ipc_framed::OutputStream;
+use codex_windows_sandbox::log_note;
+use codex_windows_sandbox::parse_policy;
+use codex_windows_sandbox::read_handle_loop;
+use codex_windows_sandbox::spawn_process_with_pipes;
+use codex_windows_sandbox::to_wide;
+use codex_windows_sandbox::PipeSpawnHandles;
+use codex_windows_sandbox::SandboxPolicy;
+use codex_windows_sandbox::StderrMode;
+use codex_windows_sandbox::StdinMode;
+use serde::Deserialize;
+use std::collections::HashMap;
+use std::ffi::c_void;
+use std::fs::File;
+use std::os::windows::io::FromRawHandle;
+use std::path::Path;
+use std::path::PathBuf;
+use std::ptr;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+use windows_sys::Win32::Foundation::CloseHandle;
+use windows_sys::Win32::Foundation::GetLastError;
+use windows_sys::Win32::Foundation::LocalFree;
+use windows_sys::Win32::Foundation::HANDLE;
+use windows_sys::Win32::Foundation::HLOCAL;
+use windows_sys::Win32::Storage::FileSystem::CreateFileW;
+use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_READ;
+use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_WRITE;
+use windows_sys::Win32::Storage::FileSystem::OPEN_EXISTING;
+use windows_sys::Win32::System::Console::ClosePseudoConsole;
+use windows_sys::Win32::System::JobObjects::AssignProcessToJobObject;
+use windows_sys::Win32::System::JobObjects::CreateJobObjectW;
+use windows_sys::Win32::System::JobObjects::JobObjectExtendedLimitInformation;
+use windows_sys::Win32::System::JobObjects::SetInformationJobObject;
+use windows_sys::Win32::System::JobObjects::JOBOBJECT_EXTENDED_LIMIT_INFORMATION;
+use windows_sys::Win32::System::JobObjects::JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+use windows_sys::Win32::System::Threading::GetExitCodeProcess;
+use windows_sys::Win32::System::Threading::GetProcessId;
+use windows_sys::Win32::System::Threading::TerminateProcess;
+use windows_sys::Win32::System::Threading::WaitForSingleObject;
+use windows_sys::Win32::System::Threading::INFINITE;
+use windows_sys::Win32::System::Threading::PROCESS_INFORMATION;
+
+#[path = "cwd_junction.rs"]
+mod cwd_junction;
+
+#[allow(dead_code)]
+#[path = "../read_acl_mutex.rs"]
+mod read_acl_mutex;
+
+#[derive(Debug, Deserialize)]
+struct RunnerRequest {
+    policy_json_or_preset: String,
+    codex_home: PathBuf,
+    real_codex_home: PathBuf,
+    cap_sids: Vec<String>,
+    command: Vec<String>,
+    cwd: PathBuf,
+    env_map: HashMap<String, String>,
+    timeout_ms: Option<u64>,
+    use_private_desktop: bool,
+    stdin_pipe: String,
+    stdout_pipe: String,
+    stderr_pipe: String,
+}
+
+const WAIT_TIMEOUT: u32 = 0x0000_0102;
+
+struct IpcSpawnedProcess {
+    log_dir: PathBuf,
+    pi: PROCESS_INFORMATION,
+    stdout_handle: HANDLE,
+    stderr_handle: HANDLE,
+    stdin_handle: Option<HANDLE>,
+    hpc_handle: Option<HANDLE>,
+}
+
+unsafe fn create_job_kill_on_close() -> Result<HANDLE> {
+    let h = CreateJobObjectW(std::ptr::null_mut(), std::ptr::null());
+    if h == 0 {
+        return Err(anyhow::anyhow!("CreateJobObjectW failed"));
+    }
+    let mut limits: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed();
+    limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+    let ok = SetInformationJobObject(
+        h,
+        JobObjectExtendedLimitInformation,
+        &mut limits as *mut _ as *mut _,
+        std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
+    );
+    if ok == 0 {
+        return Err(anyhow::anyhow!("SetInformationJobObject failed"));
+    }
+    Ok(h)
+}
+
+/// Open a named pipe created by the parent process.
+fn open_pipe(name: &str, access: u32) -> Result<HANDLE> {
+    let path = to_wide(name);
+    let handle = unsafe {
+        CreateFileW(
+            path.as_ptr(),
+            access,
+            0,
+            std::ptr::null_mut(),
+            OPEN_EXISTING,
+            0,
+            0,
+        )
+    };
+    if handle == windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE {
+        let err = unsafe { GetLastError() };
+        return Err(anyhow::anyhow!("CreateFileW failed for pipe {name}: {err}"));
+    }
+    Ok(handle)
+}
+
+fn read_request_file(req_path: &Path) -> Result<String> {
+    let content = std::fs::read_to_string(req_path)
+        .with_context(|| format!("read request file {}", req_path.display()));
+    let _ = std::fs::remove_file(req_path);
+    content
+}
+
+/// Send an error frame back to the parent process.
+fn send_error(writer: &Arc<StdMutex<File>>, code: &str, message: String) -> Result<()> {
+    let msg = FramedMessage {
+        version: 1,
+        message: Message::Error {
+            payload: ErrorPayload {
+                message,
+                code: code.to_string(),
+            },
+        },
+    };
+    if let Ok(mut guard) = writer.lock() {
+        write_frame(&mut *guard, &msg)?;
+    }
+    Ok(())
+}
+
+/// Read and validate the initial spawn request frame.
+fn read_spawn_request(
+    reader: &mut File,
+) -> Result<codex_windows_sandbox::ipc_framed::SpawnRequest> {
+    let Some(msg) = read_frame(reader)? else {
+        anyhow::bail!("runner: pipe closed before spawn_request");
+    };
+    if msg.version != 1 {
+        anyhow::bail!("runner: unsupported protocol version {}", msg.version);
+    }
+    match msg.message {
+        Message::SpawnRequest { payload } => Ok(*payload),
+        other => anyhow::bail!("runner: expected spawn_request, got {other:?}"),
+    }
+}
+
+/// Pick an effective CWD, using a junction if the ACL helper is active.
+fn effective_cwd(req_cwd: &Path, log_dir: Option<&Path>) -> PathBuf {
+    let use_junction = match read_acl_mutex::read_acl_mutex_exists() {
+        Ok(exists) => exists,
+        Err(err) => {
+            log_note(
+                &format!(
+                    "junction: read_acl_mutex_exists failed: {err}; assuming read ACL helper is running"
+                ),
+                log_dir,
+            );
+            true
+        }
+    };
+    if use_junction {
+        log_note(
+            "junction: read ACL helper running; using junction CWD",
+            log_dir,
+        );
+        cwd_junction::create_cwd_junction(req_cwd, log_dir).unwrap_or_else(|| req_cwd.to_path_buf())
+    } else {
+        req_cwd.to_path_buf()
+    }
+}
+
+fn spawn_ipc_process(
+    req: &codex_windows_sandbox::ipc_framed::SpawnRequest,
+) -> Result<IpcSpawnedProcess> {
+    let log_dir = req.codex_home.clone();
+    hide_current_user_profile_dir(req.codex_home.as_path());
+    log_note(
+        &format!(
+            "runner start cwd={} cmd={:?} real_codex_home={}",
+            req.cwd.display(),
+            req.command,
+            req.real_codex_home.display()
+        ),
+        Some(&req.codex_home),
+    );
+
+    let policy = parse_policy(&req.policy_json_or_preset).context("parse policy_json_or_preset")?;
+    if !policy.has_full_disk_read_access() {
+        anyhow::bail!(
+            "Restricted read-only access is not yet supported by the Windows sandbox backend"
+        );
+    }
+    let mut cap_psids: Vec<*mut c_void> = Vec::new();
+    for sid in &req.cap_sids {
+        let Some(psid) = (unsafe { convert_string_sid_to_sid(sid) }) else {
+            anyhow::bail!("ConvertStringSidToSidW failed for capability SID");
+        };
+        cap_psids.push(psid);
+    }
+    if cap_psids.is_empty() {
+        anyhow::bail!("runner: empty capability SID list");
+    }
+
+    let base = unsafe { get_current_token_for_restriction()? };
+    let token_res: Result<(HANDLE, *mut c_void)> = unsafe {
+        match &policy {
+            SandboxPolicy::ReadOnly { .. } => {
+                create_readonly_token_with_caps_from(base, &cap_psids)
+                    .map(|h_token| (h_token, cap_psids[0]))
+            }
+            SandboxPolicy::WorkspaceWrite { .. } => {
+                create_workspace_write_token_with_caps_from(base, &cap_psids)
+                    .map(|h_token| (h_token, cap_psids[0]))
+            }
+            SandboxPolicy::DangerFullAccess | SandboxPolicy::ExternalSandbox { .. } => {
+                unreachable!()
+            }
+        }
+    };
+    let (h_token, psid_to_use) = token_res?;
+    unsafe {
+        CloseHandle(base);
+        allow_null_device(psid_to_use);
+        for psid in &cap_psids {
+            allow_null_device(*psid);
+        }
+        for psid in cap_psids {
+            if !psid.is_null() {
+                LocalFree(psid as HLOCAL);
+            }
+        }
+    }
+
+    let effective_cwd = effective_cwd(&req.cwd, Some(log_dir.as_path()));
+    log_note(
+        &format!(
+            "runner: effective cwd={} (requested {})",
+            effective_cwd.display(),
+            req.cwd.display()
+        ),
+        Some(log_dir.as_path()),
+    );
+
+    let mut hpc_handle: Option<HANDLE> = None;
+    let (pi, stdout_handle, stderr_handle, stdin_handle) = if req.tty {
+        let (pi, conpty) = codex_windows_sandbox::spawn_conpty_process_as_user(
+            h_token,
+            &req.command,
+            &effective_cwd,
+            &req.env,
+        )?;
+        let (hpc, input_write, output_read) = conpty.into_raw();
+        hpc_handle = Some(hpc);
+        let stdin_handle = if req.stdin_open {
+            Some(input_write)
+        } else {
+            unsafe {
+                CloseHandle(input_write);
+            }
+            None
+        };
+        (
+            pi,
+            output_read,
+            windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE,
+            stdin_handle,
+        )
+    } else {
+        let stdin_mode = if req.stdin_open {
+            StdinMode::Open
+        } else {
+            StdinMode::Closed
+        };
+        let pipe_handles: PipeSpawnHandles = spawn_process_with_pipes(
+            h_token,
+            &req.command,
+            &effective_cwd,
+            &req.env,
+            stdin_mode,
+            StderrMode::Separate,
+        )?;
+        (
+            pipe_handles.process,
+            pipe_handles.stdout_read,
+            pipe_handles
+                .stderr_read
+                .unwrap_or(windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE),
+            pipe_handles.stdin_write,
+        )
+    };
+
+    unsafe {
+        CloseHandle(h_token);
+    }
+
+    Ok(IpcSpawnedProcess {
+        log_dir,
+        pi,
+        stdout_handle,
+        stderr_handle,
+        stdin_handle,
+        hpc_handle,
+    })
+}
+
+/// Stream stdout/stderr from the child into Output frames.
+fn spawn_output_reader(
+    writer: Arc<StdMutex<File>>,
+    handle: HANDLE,
+    stream: OutputStream,
+    log_dir: Option<PathBuf>,
+) -> std::thread::JoinHandle<()> {
+    read_handle_loop(handle, move |chunk| {
+        let msg = FramedMessage {
+            version: 1,
+            message: Message::Output {
+                payload: OutputPayload {
+                    data_b64: encode_bytes(chunk),
+                    stream,
+                },
+            },
+        };
+        if let Ok(mut guard) = writer.lock() {
+            if let Err(err) = write_frame(&mut *guard, &msg) {
+                log_note(
+                    &format!("runner output write failed: {err}"),
+                    log_dir.as_deref(),
+                );
+            }
+        }
+    })
+}
+
+/// Read stdin/terminate frames and forward to the child process.
+fn spawn_input_loop(
+    mut reader: File,
+    stdin_handle: Option<HANDLE>,
+    process_handle: Arc<StdMutex<Option<HANDLE>>>,
+    log_dir: Option<PathBuf>,
+) -> std::thread::JoinHandle<()> {
+    std::thread::spawn(move || {
+        loop {
+            let msg = match read_frame(&mut reader) {
+                Ok(Some(v)) => v,
+                Ok(None) => break,
+                Err(err) => {
+                    log_note(
+                        &format!("runner input read failed: {err}"),
+                        log_dir.as_deref(),
+                    );
+                    break;
+                }
+            };
+            match msg.message {
+                Message::Stdin { payload } => {
+                    let Ok(bytes) = decode_bytes(&payload.data_b64) else {
+                        continue;
+                    };
+                    if let Some(handle) = stdin_handle {
+                        let mut written: u32 = 0;
+                        unsafe {
+                            let _ = windows_sys::Win32::Storage::FileSystem::WriteFile(
+                                handle,
+                                bytes.as_ptr(),
+                                bytes.len() as u32,
+                                &mut written,
+                                ptr::null_mut(),
+                            );
+                        }
+                    }
+                }
+                Message::Terminate { .. } => {
+                    if let Ok(guard) = process_handle.lock() {
+                        if let Some(handle) = guard.as_ref() {
+                            unsafe {
+                                let _ = TerminateProcess(*handle, 1);
+                            }
+                        }
+                    }
+                }
+                Message::SpawnRequest { .. } => {}
+                Message::SpawnReady { .. } => {}
+                Message::Output { .. } => {}
+                Message::Exit { .. } => {}
+                Message::Error { .. } => {}
+            }
+        }
+        if let Some(handle) = stdin_handle {
+            unsafe {
+                CloseHandle(handle);
+            }
+        }
+    })
+}
+
+/// Entry point for the Windows command runner process.
+pub fn main() -> Result<()> {
+    let mut request_file = None;
+    let mut pipe_in = None;
+    let mut pipe_out = None;
+    let mut pipe_single = None;
+    for arg in std::env::args().skip(1) {
+        if let Some(rest) = arg.strip_prefix("--request-file=") {
+            request_file = Some(rest.to_string());
+        } else if let Some(rest) = arg.strip_prefix("--pipe-in=") {
+            pipe_in = Some(rest.to_string());
+        } else if let Some(rest) = arg.strip_prefix("--pipe-out=") {
+            pipe_out = Some(rest.to_string());
+        } else if let Some(rest) = arg.strip_prefix("--pipe=") {
+            pipe_single = Some(rest.to_string());
+        }
+    }
+    if pipe_in.is_none() && pipe_out.is_none() {
+        if let Some(single) = pipe_single {
+            pipe_in = Some(single.clone());
+            pipe_out = Some(single);
+        }
+    }
+
+    if let Some(request_file) = request_file {
+        let req_path = PathBuf::from(request_file);
+        let input = read_request_file(&req_path)?;
+        let req: RunnerRequest =
+            serde_json::from_str(&input).context("parse runner request json")?;
+        let log_dir = Some(req.codex_home.as_path());
+        hide_current_user_profile_dir(req.codex_home.as_path());
+        log_note(
+            &format!(
+                "runner start cwd={} cmd={:?} real_codex_home={}",
+                req.cwd.display(),
+                req.command,
+                req.real_codex_home.display()
+            ),
+            Some(&req.codex_home),
+        );
+
+        let policy =
+            parse_policy(&req.policy_json_or_preset).context("parse policy_json_or_preset")?;
+        if !policy.has_full_disk_read_access() {
+            anyhow::bail!(
+                "Restricted read-only access is not yet supported by the Windows sandbox backend"
+            );
+        }
+        let mut cap_psids: Vec<*mut c_void> = Vec::new();
+        for sid in &req.cap_sids {
+            let Some(psid) = (unsafe { convert_string_sid_to_sid(sid) }) else {
+                anyhow::bail!("ConvertStringSidToSidW failed for capability SID");
+            };
+            cap_psids.push(psid);
+        }
+        if cap_psids.is_empty() {
+            anyhow::bail!("runner: empty capability SID list");
+        }
+
+        let base = unsafe { get_current_token_for_restriction()? };
+        let token_res: Result<HANDLE> = unsafe {
+            match &policy {
+                SandboxPolicy::ReadOnly { .. } => {
+                    create_readonly_token_with_caps_from(base, &cap_psids)
+                }
+                SandboxPolicy::WorkspaceWrite { .. } => {
+                    create_workspace_write_token_with_caps_from(base, &cap_psids)
+                }
+                SandboxPolicy::DangerFullAccess | SandboxPolicy::ExternalSandbox { .. } => {
+                    unreachable!()
+                }
+            }
+        };
+        let h_token = token_res?;
+        unsafe {
+            CloseHandle(base);
+            for psid in &cap_psids {
+                allow_null_device(*psid);
+            }
+            for psid in cap_psids {
+                if !psid.is_null() {
+                    LocalFree(psid as HLOCAL);
+                }
+            }
+        }
+
+        let h_stdin = open_pipe(&req.stdin_pipe, FILE_GENERIC_READ)?;
+        let h_stdout = open_pipe(&req.stdout_pipe, FILE_GENERIC_WRITE)?;
+        let h_stderr = open_pipe(&req.stderr_pipe, FILE_GENERIC_WRITE)?;
+        let stdio = Some((h_stdin, h_stdout, h_stderr));
+
+        let effective_cwd = effective_cwd(&req.cwd, log_dir);
+        log_note(
+            &format!(
+                "runner: effective cwd={} (requested {})",
+                effective_cwd.display(),
+                req.cwd.display()
+            ),
+            log_dir,
+        );
+
+        let spawn_result = unsafe {
+            create_process_as_user(
+                h_token,
+                &req.command,
+                &effective_cwd,
+                &req.env_map,
+                Some(&req.codex_home),
+                stdio,
+                req.use_private_desktop,
+            )
+        };
+        let created = match spawn_result {
+            Ok(v) => v,
+            Err(err) => {
+                log_note(&format!("runner: spawn failed: {err:?}"), log_dir);
+                unsafe {
+                    CloseHandle(h_stdin);
+                    CloseHandle(h_stdout);
+                    CloseHandle(h_stderr);
+                    CloseHandle(h_token);
+                }
+                return Err(err);
+            }
+        };
+        let proc_info = created.process_info;
+
+        let h_job = unsafe { create_job_kill_on_close().ok() };
+        if let Some(job) = h_job {
+            unsafe {
+                let _ = AssignProcessToJobObject(job, proc_info.hProcess);
+            }
+        }
+
+        let wait_res = unsafe {
+            WaitForSingleObject(
+                proc_info.hProcess,
+                req.timeout_ms.map(|ms| ms as u32).unwrap_or(INFINITE),
+            )
+        };
+        let timed_out = wait_res == WAIT_TIMEOUT;
+
+        let exit_code: i32;
+        unsafe {
+            if timed_out {
+                let _ = TerminateProcess(proc_info.hProcess, 1);
+                exit_code = 128 + 64;
+            } else {
+                let mut raw_exit: u32 = 1;
+                GetExitCodeProcess(proc_info.hProcess, &mut raw_exit);
+                exit_code = raw_exit as i32;
+            }
+            if proc_info.hThread != 0 {
+                CloseHandle(proc_info.hThread);
+            }
+            if proc_info.hProcess != 0 {
+                CloseHandle(proc_info.hProcess);
+            }
+            CloseHandle(h_stdin);
+            CloseHandle(h_stdout);
+            CloseHandle(h_stderr);
+            CloseHandle(h_token);
+            if let Some(job) = h_job {
+                CloseHandle(job);
+            }
+        }
+        if exit_code != 0 {
+            eprintln!("runner child exited with code {exit_code}");
+        }
+        std::process::exit(exit_code);
+    }
+
+    let Some(pipe_in) = pipe_in else {
+        anyhow::bail!("runner: no pipe-in provided");
+    };
+    let Some(pipe_out) = pipe_out else {
+        anyhow::bail!("runner: no pipe-out provided");
+    };
+
+    let h_pipe_in = open_pipe(&pipe_in, FILE_GENERIC_READ)?;
+    let h_pipe_out = open_pipe(&pipe_out, FILE_GENERIC_WRITE)?;
+    let mut pipe_read = unsafe { File::from_raw_handle(h_pipe_in as _) };
+    let pipe_write = Arc::new(StdMutex::new(unsafe {
+        File::from_raw_handle(h_pipe_out as _)
+    }));
+
+    let req = match read_spawn_request(&mut pipe_read) {
+        Ok(v) => v,
+        Err(err) => {
+            let _ = send_error(&pipe_write, "spawn_failed", err.to_string());
+            return Err(err);
+        }
+    };
+
+    let ipc_spawn = match spawn_ipc_process(&req) {
+        Ok(value) => value,
+        Err(err) => {
+            let _ = send_error(&pipe_write, "spawn_failed", err.to_string());
+            return Err(err);
+        }
+    };
+    let log_dir = Some(ipc_spawn.log_dir.as_path());
+    let pi = ipc_spawn.pi;
+    let stdout_handle = ipc_spawn.stdout_handle;
+    let stderr_handle = ipc_spawn.stderr_handle;
+    let stdin_handle = ipc_spawn.stdin_handle;
+    let hpc_handle = ipc_spawn.hpc_handle;
+
+    let h_job = unsafe { create_job_kill_on_close().ok() };
+    if let Some(job) = h_job {
+        unsafe {
+            let _ = AssignProcessToJobObject(job, pi.hProcess);
+        }
+    }
+
+    let process_handle = Arc::new(StdMutex::new(Some(pi.hProcess)));
+
+    let msg = FramedMessage {
+        version: 1,
+        message: Message::SpawnReady {
+            payload: codex_windows_sandbox::ipc_framed::SpawnReady {
+                process_id: unsafe { GetProcessId(pi.hProcess) },
+            },
+        },
+    };
+    if let Err(err) = if let Ok(mut guard) = pipe_write.lock() {
+        write_frame(&mut *guard, &msg)
+    } else {
+        anyhow::bail!("runner spawn_ready write failed: pipe_write lock poisoned");
+    } {
+        log_note(&format!("runner spawn_ready write failed: {err}"), log_dir);
+        let _ = send_error(&pipe_write, "spawn_failed", err.to_string());
+        return Err(err);
+    }
+    let log_dir_owned = log_dir.map(|p| p.to_path_buf());
+    let out_thread = spawn_output_reader(
+        Arc::clone(&pipe_write),
+        stdout_handle,
+        OutputStream::Stdout,
+        log_dir_owned.clone(),
+    );
+    let err_thread = if stderr_handle != windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE {
+        Some(spawn_output_reader(
+            Arc::clone(&pipe_write),
+            stderr_handle,
+            OutputStream::Stderr,
+            log_dir_owned.clone(),
+        ))
+    } else {
+        None
+    };
+
+    let _input_thread = spawn_input_loop(
+        pipe_read,
+        stdin_handle,
+        Arc::clone(&process_handle),
+        log_dir_owned,
+    );
+
+    let timeout = req.timeout_ms.map(|ms| ms as u32).unwrap_or(INFINITE);
+    let wait_res = unsafe { WaitForSingleObject(pi.hProcess, timeout) };
+    let timed_out = wait_res == WAIT_TIMEOUT;
+
+    let exit_code: i32;
+    unsafe {
+        if timed_out {
+            let _ = TerminateProcess(pi.hProcess, 1);
+            exit_code = 128 + 64;
+        } else {
+            let mut raw_exit: u32 = 1;
+            GetExitCodeProcess(pi.hProcess, &mut raw_exit);
+            exit_code = raw_exit as i32;
+        }
+        if let Some(hpc) = hpc_handle {
+            ClosePseudoConsole(hpc);
+        }
+        if pi.hThread != 0 {
+            CloseHandle(pi.hThread);
+        }
+        if pi.hProcess != 0 {
+            CloseHandle(pi.hProcess);
+        }
+        if let Some(job) = h_job {
+            CloseHandle(job);
+        }
+    }
+    let _ = out_thread.join();
+    if let Some(err_thread) = err_thread {
+        let _ = err_thread.join();
+    }
+    let exit_msg = FramedMessage {
+        version: 1,
+        message: Message::Exit {
+            payload: ExitPayload {
+                exit_code,
+                timed_out,
+            },
+        },
+    };
+    if let Ok(mut guard) = pipe_write.lock() {
+        if let Err(err) = write_frame(&mut *guard, &exit_msg) {
+            log_note(&format!("runner exit write failed: {err}"), log_dir);
+        }
+    }
+
+    std::process::exit(exit_code);
+}

codex-rs/windows-sandbox-rs/src/elevated/ipc_framed.rs

@@ -0,0 +1,181 @@
+//! Framed IPC protocol used between the parent (CLI) and the elevated command runner.
+//!
+//! This module defines the JSON message schema (spawn request/ready, output, stdin,
+//! exit, error, terminate) plus length‑prefixed framing helpers for a byte stream.
+//! It is **elevated-path only**: the parent uses it to bootstrap the runner and
+//! stream unified_exec I/O over named pipes. The legacy restricted‑token path does
+//! not use this protocol, and non‑unified exec capture uses it only when running
+//! through the elevated runner.
+
+use anyhow::Result;
+use base64::engine::general_purpose::STANDARD;
+use base64::Engine as _;
+use serde::Deserialize;
+use serde::Serialize;
+use std::collections::HashMap;
+use std::io::Read;
+use std::io::Write;
+use std::path::PathBuf;
+
+/// Safety cap for a single framed message payload.
+///
+/// This is not a protocol requirement; it simply bounds memory use and rejects
+/// obviously invalid frames.
+const MAX_FRAME_LEN: usize = 8 * 1024 * 1024;
+
+/// Length-prefixed, JSON-encoded frame.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct FramedMessage {
+    pub version: u8,
+    #[serde(flatten)]
+    pub message: Message,
+}
+
+/// IPC message variants exchanged between parent and runner.
+///
+/// `SpawnRequest`, `Stdin`, and `Terminate` are parent->runner commands. `SpawnReady`,
+/// `Output`, `Exit`, and `Error` are runner->parent events/results.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum Message {
+    SpawnRequest { payload: Box<SpawnRequest> },
+    SpawnReady { payload: SpawnReady },
+    Output { payload: OutputPayload },
+    Stdin { payload: StdinPayload },
+    Exit { payload: ExitPayload },
+    Error { payload: ErrorPayload },
+    Terminate { payload: EmptyPayload },
+}
+
+/// Spawn parameters sent from parent to runner.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct SpawnRequest {
+    pub command: Vec<String>,
+    pub cwd: PathBuf,
+    pub env: HashMap<String, String>,
+    pub policy_json_or_preset: String,
+    pub sandbox_policy_cwd: PathBuf,
+    pub codex_home: PathBuf,
+    pub real_codex_home: PathBuf,
+    pub cap_sids: Vec<String>,
+    pub timeout_ms: Option<u64>,
+    pub tty: bool,
+    #[serde(default)]
+    pub stdin_open: bool,
+}
+
+/// Ack from runner after it spawns the child process.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct SpawnReady {
+    pub process_id: u32,
+}
+
+/// Output data sent from runner to parent.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct OutputPayload {
+    pub data_b64: String,
+    pub stream: OutputStream,
+}
+
+/// Output stream identifier for `OutputPayload`.
+#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum OutputStream {
+    Stdout,
+    Stderr,
+}
+
+/// Stdin bytes sent from parent to runner.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct StdinPayload {
+    pub data_b64: String,
+}
+
+/// Exit status sent from runner to parent.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ExitPayload {
+    pub exit_code: i32,
+    pub timed_out: bool,
+}
+
+/// Error payload sent when the runner fails to spawn or stream.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ErrorPayload {
+    pub message: String,
+    pub code: String,
+}
+
+/// Empty payload for control messages.
+#[derive(Debug, Serialize, Deserialize, Clone, Default)]
+pub struct EmptyPayload {}
+
+/// Base64-encode raw bytes for IPC payloads.
+pub fn encode_bytes(data: &[u8]) -> String {
+    STANDARD.encode(data)
+}
+
+/// Decode base64 payload data into raw bytes.
+pub fn decode_bytes(data: &str) -> Result<Vec<u8>> {
+    Ok(STANDARD.decode(data.as_bytes())?)
+}
+
+/// Write a length-prefixed JSON frame.
+pub fn write_frame<W: Write>(mut writer: W, msg: &FramedMessage) -> Result<()> {
+    let payload = serde_json::to_vec(msg)?;
+    if payload.len() > MAX_FRAME_LEN {
+        anyhow::bail!("frame too large: {}", payload.len());
+    }
+    let len = payload.len() as u32;
+    writer.write_all(&len.to_le_bytes())?;
+    writer.write_all(&payload)?;
+    writer.flush()?;
+    Ok(())
+}
+
+/// Read a length-prefixed JSON frame; returns `Ok(None)` on EOF.
+pub fn read_frame<R: Read>(mut reader: R) -> Result<Option<FramedMessage>> {
+    let mut len_buf = [0u8; 4];
+    match reader.read_exact(&mut len_buf) {
+        Ok(()) => {}
+        Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
+        Err(err) => return Err(err.into()),
+    }
+    let len = u32::from_le_bytes(len_buf) as usize;
+    if len > MAX_FRAME_LEN {
+        anyhow::bail!("frame too large: {}", len);
+    }
+    let mut payload = vec![0u8; len];
+    reader.read_exact(&mut payload)?;
+    let msg: FramedMessage = serde_json::from_slice(&payload)?;
+    Ok(Some(msg))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn framed_round_trip() {
+        let msg = FramedMessage {
+            version: 1,
+            message: Message::Output {
+                payload: OutputPayload {
+                    data_b64: encode_bytes(b"hello"),
+                    stream: OutputStream::Stdout,
+                },
+            },
+        };
+        let mut buf = Vec::new();
+        write_frame(&mut buf, &msg).expect("write");
+        let decoded = read_frame(buf.as_slice()).expect("read").expect("some");
+        assert_eq!(decoded.version, 1);
+        match decoded.message {
+            Message::Output { payload } => {
+                assert_eq!(payload.stream, OutputStream::Stdout);
+                let data = decode_bytes(&payload.data_b64).expect("decode");
+                assert_eq!(data, b"hello");
+            }
+            other => panic!("unexpected message: {other:?}"),
+        }
+    }
+}

codex-rs/windows-sandbox-rs/src/elevated/runner_pipe.rs

@@ -0,0 +1,111 @@
+//! Named pipe helpers for the elevated Windows sandbox runner.
+//!
+//! This module generates paired pipe names, creates server‑side pipes with permissive
+//! ACLs, and waits for the runner to connect. It is **elevated-path only** and is
+//! used by the parent to establish the IPC channel for both unified_exec sessions
+//! and elevated capture. The legacy restricted‑token path spawns the child directly
+//! and does not use these helpers.
+
+use crate::helper_materialization::HelperExecutable;
+use crate::helper_materialization::resolve_helper_for_launch;
+use crate::winutil::resolve_sid;
+use crate::winutil::string_from_sid_bytes;
+use crate::winutil::to_wide;
+use rand::Rng;
+use rand::SeedableRng;
+use rand::rngs::SmallRng;
+use std::io;
+use std::path::Path;
+use std::path::PathBuf;
+use std::ptr;
+use windows_sys::Win32::Foundation::GetLastError;
+use windows_sys::Win32::Foundation::HANDLE;
+use windows_sys::Win32::Security::Authorization::ConvertStringSecurityDescriptorToSecurityDescriptorW;
+use windows_sys::Win32::Security::PSECURITY_DESCRIPTOR;
+use windows_sys::Win32::Security::SECURITY_ATTRIBUTES;
+use windows_sys::Win32::System::Pipes::ConnectNamedPipe;
+use windows_sys::Win32::System::Pipes::CreateNamedPipeW;
+use windows_sys::Win32::System::Pipes::PIPE_READMODE_BYTE;
+use windows_sys::Win32::System::Pipes::PIPE_TYPE_BYTE;
+use windows_sys::Win32::System::Pipes::PIPE_WAIT;
+
+/// PIPE_ACCESS_INBOUND (win32 constant), not exposed in windows-sys 0.52.
+pub const PIPE_ACCESS_INBOUND: u32 = 0x0000_0001;
+/// PIPE_ACCESS_OUTBOUND (win32 constant), not exposed in windows-sys 0.52.
+pub const PIPE_ACCESS_OUTBOUND: u32 = 0x0000_0002;
+
+/// Resolves the elevated command runner path, preferring the copied helper under
+/// `.sandbox-bin` and falling back to the legacy sibling lookup when needed.
+pub fn find_runner_exe(codex_home: &Path, log_dir: Option<&Path>) -> PathBuf {
+    resolve_helper_for_launch(HelperExecutable::CommandRunner, codex_home, log_dir)
+}
+
+/// Generates a unique named-pipe path used to communicate with the runner process.
+pub fn pipe_pair() -> (String, String) {
+    let mut rng = SmallRng::from_entropy();
+    let base = format!(r"\\.\pipe\codex-runner-{:x}", rng.gen::<u128>());
+    (format!("{base}-in"), format!("{base}-out"))
+}
+
+/// Creates a named pipe whose DACL only allows the sandbox user to connect.
+pub fn create_named_pipe(name: &str, access: u32, sandbox_username: &str) -> io::Result<HANDLE> {
+    let sandbox_sid = resolve_sid(sandbox_username)
+        .map_err(|err| io::Error::new(io::ErrorKind::PermissionDenied, err.to_string()))?;
+    let sandbox_sid = string_from_sid_bytes(&sandbox_sid)
+        .map_err(|err| io::Error::new(io::ErrorKind::PermissionDenied, err))?;
+    let sddl = to_wide(format!("D:(A;;GA;;;{sandbox_sid})"));
+    let mut sd: PSECURITY_DESCRIPTOR = ptr::null_mut();
+    let ok = unsafe {
+        ConvertStringSecurityDescriptorToSecurityDescriptorW(
+            sddl.as_ptr(),
+            1, // SDDL_REVISION_1
+            &mut sd,
+            ptr::null_mut(),
+        )
+    };
+    if ok == 0 {
+        return Err(io::Error::from_raw_os_error(unsafe {
+            GetLastError() as i32
+        }));
+    }
+    let mut sa = SECURITY_ATTRIBUTES {
+        nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
+        lpSecurityDescriptor: sd,
+        bInheritHandle: 0,
+    };
+    let wide = to_wide(name);
+    let h = unsafe {
+        CreateNamedPipeW(
+            wide.as_ptr(),
+            access,
+            PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
+            1,
+            65536,
+            65536,
+            0,
+            &mut sa as *mut SECURITY_ATTRIBUTES,
+        )
+    };
+    if h == 0 || h == windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE {
+        return Err(io::Error::from_raw_os_error(unsafe {
+            GetLastError() as i32
+        }));
+    }
+    Ok(h)
+}
+
+/// Waits for the runner to connect to a parent-created server pipe.
+///
+/// This is parent-side only: the runner opens the pipe with `CreateFileW`, while the
+/// parent calls `ConnectNamedPipe` and tolerates the already-connected case.
+pub fn connect_pipe(h: HANDLE) -> io::Result<()> {
+    let ok = unsafe { ConnectNamedPipe(h, ptr::null_mut()) };
+    if ok == 0 {
+        let err = unsafe { GetLastError() };
+        const ERROR_PIPE_CONNECTED: u32 = 535;
+        if err != ERROR_PIPE_CONNECTED {
+            return Err(io::Error::from_raw_os_error(err as i32));
+        }
+    }
+    Ok(())
+}

codex-rs/windows-sandbox-rs/src/elevated_impl.rs

@@ -17,6 +17,8 @@ mod windows_impl {
     use crate::policy::SandboxPolicy;
     use crate::token::convert_string_sid_to_sid;
     use crate::winutil::quote_windows_arg;
+    use crate::winutil::resolve_sid;
+    use crate::winutil::string_from_sid_bytes;
     use crate::winutil::to_wide;
     use anyhow::Result;
     use rand::rngs::SmallRng;
@@ -123,10 +125,9 @@ mod windows_impl {
         format!(r"\\.\pipe\codex-runner-{:x}-{}", rng.gen::<u128>(), suffix)
     }
 
-    /// Creates a named pipe with permissive ACLs so the sandbox user can connect.
-    fn create_named_pipe(name: &str, access: u32) -> io::Result<HANDLE> {
-        // Allow sandbox users to connect by granting Everyone full access on the pipe.
-        let sddl = to_wide("D:(A;;GA;;;WD)");
+    /// Creates a named pipe whose DACL only allows the sandbox user to connect.
+    fn create_named_pipe(name: &str, access: u32, sandbox_sid: &str) -> io::Result<HANDLE> {
+        let sddl = to_wide(format!("D:(A;;GA;;;{sandbox_sid})"));
         let mut sd: PSECURITY_DESCRIPTOR = ptr::null_mut();
         let ok = unsafe {
             ConvertStringSecurityDescriptorToSecurityDescriptorW(
@@ -228,6 +229,11 @@ mod windows_impl {
         log_start(&command, logs_base_dir);
         let sandbox_creds =
             require_logon_sandbox_creds(&policy, sandbox_policy_cwd, cwd, &env_map, codex_home)?;
+        let sandbox_sid = resolve_sid(&sandbox_creds.username).map_err(|err: anyhow::Error| {
+            io::Error::new(io::ErrorKind::PermissionDenied, err.to_string())
+        })?;
+        let sandbox_sid = string_from_sid_bytes(&sandbox_sid)
+            .map_err(|err| io::Error::new(io::ErrorKind::PermissionDenied, err))?;
         // Build capability SID for ACL grants.
         if matches!(
             &policy,
@@ -272,14 +278,17 @@ mod windows_impl {
         let h_stdin_pipe = create_named_pipe(
             &stdin_name,
             PIPE_ACCESS_DUPLEX | PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
+            &sandbox_sid,
         )?;
         let h_stdout_pipe = create_named_pipe(
             &stdout_name,
             PIPE_ACCESS_DUPLEX | PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
+            &sandbox_sid,
         )?;
         let h_stderr_pipe = create_named_pipe(
             &stderr_name,
             PIPE_ACCESS_DUPLEX | PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
+            &sandbox_sid,
         )?;
 
         // Launch runner as sandbox user via CreateProcessWithLogonW.

codex-rs/windows-sandbox-rs/src/lib.rs

@@ -24,6 +24,14 @@ windows_modules!(
     workspace_acl
 );
 
+#[cfg(target_os = "windows")]
+#[path = "conpty/mod.rs"]
+mod conpty;
+
+#[cfg(target_os = "windows")]
+#[path = "elevated/ipc_framed.rs"]
+pub mod ipc_framed;
+
 #[cfg(target_os = "windows")]
 #[path = "setup_orchestrator.rs"]
 mod setup;
@@ -36,6 +44,7 @@ mod setup_error;
 
 #[cfg(target_os = "windows")]
 pub use acl::add_deny_write_ace;
+
 #[cfg(target_os = "windows")]
 pub use acl::allow_null_device;
 #[cfg(target_os = "windows")]
@@ -55,6 +64,8 @@ pub use cap::load_or_create_cap_sids;
 #[cfg(target_os = "windows")]
 pub use cap::workspace_cap_sid_for_cwd;
 #[cfg(target_os = "windows")]
+pub use conpty::spawn_conpty_process_as_user;
+#[cfg(target_os = "windows")]
 pub use dpapi::protect as dpapi_protect;
 #[cfg(target_os = "windows")]
 pub use dpapi::unprotect as dpapi_unprotect;
@@ -83,6 +94,16 @@ pub use policy::SandboxPolicy;
 #[cfg(target_os = "windows")]
 pub use process::create_process_as_user;
 #[cfg(target_os = "windows")]
+pub use process::read_handle_loop;
+#[cfg(target_os = "windows")]
+pub use process::spawn_process_with_pipes;
+#[cfg(target_os = "windows")]
+pub use process::PipeSpawnHandles;
+#[cfg(target_os = "windows")]
+pub use process::StderrMode;
+#[cfg(target_os = "windows")]
+pub use process::StdinMode;
+#[cfg(target_os = "windows")]
 pub use setup::run_elevated_setup;
 #[cfg(target_os = "windows")]
 pub use setup::run_setup_refresh;

codex-rs/windows-sandbox-rs/src/process.rs

@@ -8,15 +8,19 @@ use anyhow::Result;
 use std::collections::HashMap;
 use std::ffi::c_void;
 use std::path::Path;
+use std::ptr;
 use windows_sys::Win32::Foundation::GetLastError;
+use windows_sys::Win32::Foundation::CloseHandle;
 use windows_sys::Win32::Foundation::SetHandleInformation;
 use windows_sys::Win32::Foundation::HANDLE;
 use windows_sys::Win32::Foundation::HANDLE_FLAG_INHERIT;
 use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
+use windows_sys::Win32::Storage::FileSystem::ReadFile;
 use windows_sys::Win32::System::Console::GetStdHandle;
 use windows_sys::Win32::System::Console::STD_ERROR_HANDLE;
 use windows_sys::Win32::System::Console::STD_INPUT_HANDLE;
 use windows_sys::Win32::System::Console::STD_OUTPUT_HANDLE;
+use windows_sys::Win32::System::Pipes::CreatePipe;
 use windows_sys::Win32::System::Threading::CreateProcessAsUserW;
 use windows_sys::Win32::System::Threading::CREATE_UNICODE_ENVIRONMENT;
 use windows_sys::Win32::System::Threading::PROCESS_INFORMATION;
@@ -153,3 +157,141 @@ pub unsafe fn create_process_as_user(
         _desktop: desktop,
     })
 }
+
+/// Controls whether the child's stdin handle is kept open for writing.
+#[allow(dead_code)]
+pub enum StdinMode {
+    Closed,
+    Open,
+}
+
+/// Controls how stderr is wired for a pipe-spawned process.
+#[allow(dead_code)]
+pub enum StderrMode {
+    MergeStdout,
+    Separate,
+}
+
+/// Handles returned by `spawn_process_with_pipes`.
+#[allow(dead_code)]
+pub struct PipeSpawnHandles {
+    pub process: PROCESS_INFORMATION,
+    pub stdin_write: Option<HANDLE>,
+    pub stdout_read: HANDLE,
+    pub stderr_read: Option<HANDLE>,
+}
+
+/// Spawns a process with anonymous pipes and returns the relevant handles.
+pub fn spawn_process_with_pipes(
+    h_token: HANDLE,
+    argv: &[String],
+    cwd: &Path,
+    env_map: &HashMap<String, String>,
+    stdin_mode: StdinMode,
+    stderr_mode: StderrMode,
+) -> Result<PipeSpawnHandles> {
+    let mut in_r: HANDLE = 0;
+    let mut in_w: HANDLE = 0;
+    let mut out_r: HANDLE = 0;
+    let mut out_w: HANDLE = 0;
+    let mut err_r: HANDLE = 0;
+    let mut err_w: HANDLE = 0;
+    unsafe {
+        if CreatePipe(&mut in_r, &mut in_w, ptr::null_mut(), 0) == 0 {
+            return Err(anyhow!("CreatePipe stdin failed: {}", GetLastError()));
+        }
+        if CreatePipe(&mut out_r, &mut out_w, ptr::null_mut(), 0) == 0 {
+            CloseHandle(in_r);
+            CloseHandle(in_w);
+            return Err(anyhow!("CreatePipe stdout failed: {}", GetLastError()));
+        }
+        if matches!(stderr_mode, StderrMode::Separate)
+            && CreatePipe(&mut err_r, &mut err_w, ptr::null_mut(), 0) == 0
+        {
+            CloseHandle(in_r);
+            CloseHandle(in_w);
+            CloseHandle(out_r);
+            CloseHandle(out_w);
+            return Err(anyhow!("CreatePipe stderr failed: {}", GetLastError()));
+        }
+    }
+
+    let stderr_handle = match stderr_mode {
+        StderrMode::MergeStdout => out_w,
+        StderrMode::Separate => err_w,
+    };
+
+    let stdio = Some((in_r, out_w, stderr_handle));
+    let spawn_result =
+        unsafe { create_process_as_user(h_token, argv, cwd, env_map, None, stdio, false) };
+    let created = match spawn_result {
+        Ok(v) => v,
+        Err(err) => {
+            unsafe {
+                CloseHandle(in_r);
+                CloseHandle(in_w);
+                CloseHandle(out_r);
+                CloseHandle(out_w);
+                if matches!(stderr_mode, StderrMode::Separate) {
+                    CloseHandle(err_r);
+                    CloseHandle(err_w);
+                }
+            }
+            return Err(err);
+        }
+    };
+    let pi = created.process_info;
+
+    unsafe {
+        CloseHandle(in_r);
+        CloseHandle(out_w);
+        if matches!(stderr_mode, StderrMode::Separate) {
+            CloseHandle(err_w);
+        }
+        if matches!(stdin_mode, StdinMode::Closed) {
+            CloseHandle(in_w);
+        }
+    }
+
+    Ok(PipeSpawnHandles {
+        process: pi,
+        stdin_write: match stdin_mode {
+            StdinMode::Open => Some(in_w),
+            StdinMode::Closed => None,
+        },
+        stdout_read: out_r,
+        stderr_read: match stderr_mode {
+            StderrMode::Separate => Some(err_r),
+            StderrMode::MergeStdout => None,
+        },
+    })
+}
+
+/// Reads a HANDLE until EOF and invokes `on_chunk` for each read.
+pub fn read_handle_loop<F>(handle: HANDLE, mut on_chunk: F) -> std::thread::JoinHandle<()>
+where
+    F: FnMut(&[u8]) + Send + 'static,
+{
+    std::thread::spawn(move || {
+        let mut buf = [0u8; 8192];
+        loop {
+            let mut read_bytes: u32 = 0;
+            let ok = unsafe {
+                ReadFile(
+                    handle,
+                    buf.as_mut_ptr(),
+                    buf.len() as u32,
+                    &mut read_bytes,
+                    ptr::null_mut(),
+                )
+            };
+            if ok == 0 || read_bytes == 0 {
+                break;
+            }
+            on_chunk(&buf[..read_bytes as usize]);
+        }
+        unsafe {
+            CloseHandle(handle);
+        }
+    })
+}

codex-rs/windows-sandbox-rs/src/sandbox_utils.rs

@@ -0,0 +1,67 @@
+//! Shared helper utilities for Windows sandbox setup.
+//!
+//! These helpers centralize small pieces of setup logic used across both legacy and
+//! elevated paths, including unified_exec sessions and capture flows. They cover
+//! codex home directory creation and git safe.directory injection so sandboxed
+//! users can run git inside a repo owned by the primary user.
+
+use anyhow::Result;
+use std::collections::HashMap;
+use std::path::Path;
+use std::path::PathBuf;
+
+/// Walk upward from `start` to locate the git worktree root (supports gitfile redirects).
+fn find_git_root(start: &Path) -> Option<PathBuf> {
+    let mut cur = dunce::canonicalize(start).ok()?;
+    loop {
+        let marker = cur.join(".git");
+        if marker.is_dir() {
+            return Some(cur);
+        }
+        if marker.is_file() {
+            if let Ok(txt) = std::fs::read_to_string(&marker) {
+                if let Some(rest) = txt.trim().strip_prefix("gitdir:") {
+                    let gitdir = rest.trim();
+                    let resolved = if Path::new(gitdir).is_absolute() {
+                        PathBuf::from(gitdir)
+                    } else {
+                        cur.join(gitdir)
+                    };
+                    return resolved.parent().map(|p| p.to_path_buf()).or(Some(cur));
+                }
+            }
+            return Some(cur);
+        }
+        let parent = cur.parent()?;
+        if parent == cur {
+            return None;
+        }
+        cur = parent.to_path_buf();
+    }
+}
+
+/// Ensure the sandbox codex home directory exists.
+pub fn ensure_codex_home_exists(p: &Path) -> Result<()> {
+    std::fs::create_dir_all(p)?;
+    Ok(())
+}
+
+/// Adds a git safe.directory entry to the environment when running inside a repository.
+/// git will not otherwise allow the Sandbox user to run git commands on the repo directory
+/// which is owned by the primary user.
+pub fn inject_git_safe_directory(env_map: &mut HashMap<String, String>, cwd: &Path) {
+    if let Some(git_root) = find_git_root(cwd) {
+        let mut cfg_count: usize = env_map
+            .get("GIT_CONFIG_COUNT")
+            .and_then(|v| v.parse::<usize>().ok())
+            .unwrap_or(0);
+        let git_path = git_root.to_string_lossy().replace("\\\\", "/");
+        env_map.insert(
+            format!("GIT_CONFIG_KEY_{cfg_count}"),
+            "safe.directory".to_string(),
+        );
+        env_map.insert(format!("GIT_CONFIG_VALUE_{cfg_count}"), git_path);
+        cfg_count += 1;
+        env_map.insert("GIT_CONFIG_COUNT".to_string(), cfg_count.to_string());
+    }
+}

codex-rs/windows-sandbox-rs/src/winutil.rs

@@ -1,7 +1,15 @@
+use anyhow::Result;
 use std::ffi::OsStr;
 use std::os::windows::ffi::OsStrExt;
+use windows_sys::Win32::Foundation::ERROR_INSUFFICIENT_BUFFER;
+use windows_sys::Win32::Foundation::GetLastError;
 use windows_sys::Win32::Foundation::LocalFree;
 use windows_sys::Win32::Foundation::HLOCAL;
+use windows_sys::Win32::Security::Authorization::ConvertStringSidToSidW;
+use windows_sys::Win32::Security::CopySid;
+use windows_sys::Win32::Security::GetLengthSid;
+use windows_sys::Win32::Security::LookupAccountNameW;
+use windows_sys::Win32::Security::SID_NAME_USE;
 use windows_sys::Win32::System::Diagnostics::Debug::FormatMessageW;
 use windows_sys::Win32::System::Diagnostics::Debug::FORMAT_MESSAGE_ALLOCATE_BUFFER;
 use windows_sys::Win32::System::Diagnostics::Debug::FORMAT_MESSAGE_FROM_SYSTEM;
@@ -102,3 +110,83 @@ pub fn string_from_sid_bytes(sid: &[u8]) -> Result<String, String> {
         Ok(out)
     }
 }
+
+const SID_ADMINISTRATORS: &str = "S-1-5-32-544";
+const SID_USERS: &str = "S-1-5-32-545";
+const SID_AUTHENTICATED_USERS: &str = "S-1-5-11";
+const SID_EVERYONE: &str = "S-1-1-0";
+const SID_SYSTEM: &str = "S-1-5-18";
+
+pub fn resolve_sid(name: &str) -> Result<Vec<u8>> {
+    if let Some(sid_str) = well_known_sid_str(name) {
+        return sid_bytes_from_string(sid_str);
+    }
+    let name_w = to_wide(OsStr::new(name));
+    let mut sid_buffer = vec![0u8; 68];
+    let mut sid_len: u32 = sid_buffer.len() as u32;
+    let mut domain: Vec<u16> = Vec::new();
+    let mut domain_len: u32 = 0;
+    let mut use_type: SID_NAME_USE = 0;
+    loop {
+        let ok = unsafe {
+            LookupAccountNameW(
+                std::ptr::null(),
+                name_w.as_ptr(),
+                sid_buffer.as_mut_ptr() as *mut std::ffi::c_void,
+                &mut sid_len,
+                domain.as_mut_ptr(),
+                &mut domain_len,
+                &mut use_type,
+            )
+        };
+        if ok != 0 {
+            sid_buffer.truncate(sid_len as usize);
+            return Ok(sid_buffer);
+        }
+        let err = unsafe { GetLastError() };
+        if err == ERROR_INSUFFICIENT_BUFFER {
+            sid_buffer.resize(sid_len as usize, 0);
+            domain.resize(domain_len as usize, 0);
+            continue;
+        }
+        return Err(anyhow::anyhow!("LookupAccountNameW failed for {name}: {err}"));
+    }
+}
+
+fn well_known_sid_str(name: &str) -> Option<&'static str> {
+    match name {
+        "Administrators" => Some(SID_ADMINISTRATORS),
+        "Users" => Some(SID_USERS),
+        "Authenticated Users" => Some(SID_AUTHENTICATED_USERS),
+        "Everyone" => Some(SID_EVERYONE),
+        "SYSTEM" => Some(SID_SYSTEM),
+        _ => None,
+    }
+}
+
+fn sid_bytes_from_string(sid_str: &str) -> Result<Vec<u8>> {
+    let sid_w = to_wide(OsStr::new(sid_str));
+    let mut psid: *mut std::ffi::c_void = std::ptr::null_mut();
+    if unsafe { ConvertStringSidToSidW(sid_w.as_ptr(), &mut psid) } == 0 {
+        return Err(anyhow::anyhow!(
+            "ConvertStringSidToSidW failed for {sid_str}: {}",
+            unsafe { GetLastError() }
+        ));
+    }
+    let sid_len = unsafe { GetLengthSid(psid) };
+    if sid_len == 0 {
+        unsafe {
+            LocalFree(psid as _);
+        }
+        return Err(anyhow::anyhow!("GetLengthSid failed for {sid_str}"));
+    }
+    let mut out = vec![0u8; sid_len as usize];
+    let ok = unsafe { CopySid(sid_len, out.as_mut_ptr() as *mut std::ffi::c_void, psid) };
+    unsafe {
+        LocalFree(psid as _);
+    }
+    if ok == 0 {
+        return Err(anyhow::anyhow!("CopySid failed for {sid_str}"));
+    }
+    Ok(out)
+}