Add managed command checkpoints requiring fresh human approval (#58045)

## Why

Managed command policies need to require a human decision independently of automatic review, hook allowances, and saved approvals.

## What changed

- Support `decision = "prompt-user"` in managed `requirements.toml` prefix rules, requiring a fresh, one-time human decision for matching commands.
- Add experimental `humanCommand` approval requests for clients advertising both `experimentalApi` and `requiredHumanApproval`. Accept only `accept`, `decline`, or `cancel`; reject commands when no capable client is available.
- Restrict responses to capable clients that received the request, and replay pending prompts to compatible clients on reconnect.
- Enable the capability in the TUI and update protocol schemas and client bindings.

## Testing

Add coverage for fresh decisions overriding saved allowances, hook allowances, and automatic review, plus child-thread approval routing and rejection of responses from incompatible or unrelated clients.

GitOrigin-RevId: 0b95ff6aab7f8a04b7298c8a4329d1fddf6a603e

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

@@ -751,7 +751,8 @@ pub(crate) async fn apply_bespoke_event_handling(
                         Some(shlex_join(&command)),
                         Some(completion_item.cwd.clone()),
                         Some(command_actions),
-                        Some(completion_item),
+                        (kind != codex_protocol::approvals::ExecApprovalKind::HumanCommand)
+                            .then_some(completion_item),
                     ),
                 };
             if approval_starts_command && let Some(completion_item) = completion_item.as_ref() {

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

@@ -1,4 +1,4 @@
-//! Exercises thread cleanup when startup or child creation outlives the client connection.
+//! Exercises client attachment, approval routing, and cleanup across thread creation.
 
 use codex_app_server_protocol::TurnStartResponse;
 use codex_features::Feature;
@@ -24,8 +24,11 @@ use anyhow::Result;
 use app_test_support::MockResponsesConfig;
 use app_test_support::create_mock_responses_server_sequence_unchecked;
 use codex_app_server_protocol::InitializeResponse;
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::ThreadClosedNotification;
 use codex_app_server_protocol::ThreadLoadedListResponse;
 use codex_app_server_protocol::ThreadStartResponse;
@@ -510,6 +513,233 @@ async fn disconnected_v2_child_lifecycle(unload_delay_secs: u64, ephemeral: bool
     result
 }
 
+#[tokio::test]
+async fn child_human_approval_survives_attachment_and_rejects_unrelated_clients() -> Result<()> {
+    let (release_spawn_completion, spawn_completion_gate) = oneshot::channel();
+    let (server, _completions) = start_streaming_sse_server(vec![
+        vec![
+            StreamingSseChunk {
+                gate: None,
+                body: responses::sse(vec![
+                    responses::ev_response_created("spawn"),
+                    responses::ev_function_call_with_namespace(
+                        "spawn",
+                        "collaboration",
+                        "spawn_agent",
+                        &json!({"task_name": "child", "message": "Run echo", "fork_turns": "none"})
+                            .to_string(),
+                    ),
+                ]),
+            },
+            StreamingSseChunk {
+                gate: Some(spawn_completion_gate),
+                body: responses::sse(vec![responses::ev_completed("spawn")]),
+            },
+        ],
+        vec![StreamingSseChunk {
+            gate: None,
+            body: responses::sse(vec![
+                responses::ev_response_created("command"),
+                responses::ev_function_call(
+                    "child-command",
+                    "exec_command",
+                    &json!({"cmd": "echo human-approved"}).to_string(),
+                ),
+                responses::ev_completed("command"),
+            ]),
+        }],
+    ])
+    .await;
+    let home = TempDir::new()?;
+    std::fs::write(
+        home.path().join("requirements.toml"),
+        r#"
+[features]
+unified_exec = false
+shell_tool = true
+
+[[rules.prefix_rules]]
+pattern = [{ token = "echo" }]
+decision = "prompt-user"
+"#,
+    )?;
+    MockResponsesConfig::new(server.uri())
+        .with_approval_policy("on-request")
+        .with_sandbox_mode("workspace-write")
+        .with_root_config("approvals_reviewer = \"auto_review\"")
+        .enable_feature(Feature::Collab)
+        .enable_feature(Feature::MultiAgentV2)
+        .write(home.path())?;
+    let config = Arc::new(
+        ConfigBuilder::default()
+            .codex_home(home.path().to_path_buf())
+            .build()
+            .await?,
+    );
+    let auth =
+        AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false)
+            .await?;
+    let (processor, mut outgoing) = build_test_processor(config, auth).await;
+    let mut created = processor.thread_created_receiver();
+    let incompatible_id = ConnectionId(1);
+    let capable_id = ConnectionId(2);
+    let unrelated_id = ConnectionId(3);
+    let incompatible = Arc::new(ConnectionSessionState::new(ConnectionOrigin::WebSocket));
+    let capable = Arc::new(ConnectionSessionState::new(ConnectionOrigin::WebSocket));
+    let unrelated = Arc::new(ConnectionSessionState::new(ConnectionOrigin::WebSocket));
+    let result: Result<()> = async {
+        for (id, session, required_human_approval) in [
+            (incompatible_id, &incompatible, false),
+            (capable_id, &capable, true),
+        ] {
+            send_request(
+                &processor,
+                id,
+                session,
+                json!({"id": 1, "method": "initialize", "params": {
+                    "clientInfo": {"name": "attachment-test", "version": "1"},
+                    "capabilities": {
+                        "experimentalApi": true,
+                        "requiredHumanApproval": required_human_approval
+                    }
+                }}),
+            )
+            .await;
+            let _: InitializeResponse =
+                read_response_from(&mut outgoing, id, /*request_id*/ 1).await;
+            processor
+                .connection_initialized(id, session.request_attestation())
+                .await;
+        }
+        send_request(
+            &processor,
+            incompatible_id,
+            &incompatible,
+            json!({"id": 2, "method": "thread/start", "params": {}}),
+        )
+        .await;
+        let parent: ThreadStartResponse =
+            read_response_from(&mut outgoing, incompatible_id, /*request_id*/ 2).await;
+        send_request(
+            &processor,
+            incompatible_id,
+            &incompatible,
+            json!({"id": 3, "method": "turn/start", "params": {
+                "threadId": parent.thread.id,
+                "input": [{"type": "text", "text": "Spawn a child to run echo"}]
+            }}),
+        )
+        .await;
+        let _: TurnStartResponse =
+            read_response_from(&mut outgoing, incompatible_id, /*request_id*/ 3).await;
+        let child_id = timeout(Duration::from_secs(/*secs*/ 10), created.recv()).await??;
+
+        while outgoing.try_recv().is_ok() {}
+        // Attach the initial client set before draining the child's queued approval.
+        processor
+            .try_attach_thread_listener(child_id, vec![incompatible_id, capable_id])
+            .await;
+        let request = timeout(Duration::from_secs(/*secs*/ 10), async {
+            loop {
+                if let OutgoingEnvelope::ToConnection {
+                    connection_id,
+                    message: OutgoingMessage::Request(request),
+                    ..
+                } = outgoing.recv().await.context("outgoing channel closed")?
+                {
+                    assert_eq!(connection_id, capable_id);
+                    break anyhow::Ok(request);
+                }
+            }
+        })
+        .await
+        .context("capable client did not receive the child's approval during attachment")??;
+        let ServerRequest::CommandExecutionRequestApproval { request_id, params } = &request else {
+            anyhow::bail!("expected human approval, got {request:?}");
+        };
+        assert_eq!(
+            (&params.thread_id, params.kind),
+            (
+                &child_id.to_string(),
+                codex_app_server_protocol::CommandExecutionApprovalKind::HumanCommand,
+            ),
+        );
+
+        // Initialize a capable client after auto-attachment. Capability alone must not
+        // let it approve or reject this request before it subscribes and receives it.
+        send_request(
+            &processor,
+            unrelated_id,
+            &unrelated,
+            json!({"id": 1, "method": "initialize", "params": {
+                "clientInfo": {"name": "unrelated-client", "version": "1"},
+                "capabilities": {"experimentalApi": true, "requiredHumanApproval": true}
+            }}),
+        )
+        .await;
+        let _: InitializeResponse =
+            read_response_from(&mut outgoing, unrelated_id, /*request_id*/ 1).await;
+        processor
+            .connection_initialized(unrelated_id, unrelated.request_attestation())
+            .await;
+        for connection_id in [incompatible_id, unrelated_id] {
+            processor
+                .process_response(
+                    connection_id,
+                    JSONRPCResponse {
+                        id: request_id.clone(),
+                        result: json!({"decision": "accept"}),
+                    },
+                )
+                .await;
+            processor
+                .process_error(
+                    connection_id,
+                    JSONRPCError {
+                        id: request_id.clone(),
+                        error: crate::error_code::internal_error("unsupported request"),
+                    },
+                )
+                .await;
+            assert_eq!(
+                processor
+                    .outgoing
+                    .pending_requests_for_thread(child_id)
+                    .await,
+                vec![request.clone()],
+                "unrouted responses and errors must leave the request pending",
+            );
+        }
+
+        // Routing ends at the human decision. Command execution and replay have
+        // their own coverage; declining here needs no Linux sandbox executable.
+        processor
+            .process_response(
+                capable_id,
+                JSONRPCResponse {
+                    id: request_id.clone(),
+                    result: json!({"decision": "decline"}),
+                },
+            )
+            .await;
+        assert!(
+            processor
+                .outgoing
+                .pending_requests_for_thread(child_id)
+                .await
+                .is_empty()
+        );
+        Ok(())
+    }
+    .await;
+    drop(release_spawn_completion);
+    drop(outgoing);
+    processor.clear_runtime_references();
+    processor.shutdown_threads().await;
+    server.shutdown().await;
+    result
+}
+
 async fn send_request(
     processor: &Arc<MessageProcessor>,
     connection_id: ConnectionId,

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

@@ -136,6 +136,7 @@ pub(crate) enum OutgoingEnvelope {
 pub(crate) struct OutgoingMessageSender {
     verification_auth: OnceLock<Arc<codex_login::AuthManager>>,
     verification_connections: Mutex<HashSet<ConnectionId>>,
+    human_approval_connections: Mutex<HashSet<ConnectionId>>,
     next_server_request_id: AtomicI64,
     sender: mpsc::Sender<OutgoingEnvelope>,
     process_reply_sender: mpsc::Sender<OutgoingEnvelope>,
@@ -155,6 +156,7 @@ pub(crate) struct ThreadScopedOutgoingMessageSender {
 }
 
 struct PendingCallbackEntry {
+    human_approval_recipients: HashSet<ConnectionId>,
     verification_owner: Option<ConnectionId>,
     verification_auth_revision: Option<u64>,
     verification_identity: Option<user_verification_auth::Identity>,
@@ -259,6 +261,7 @@ impl OutgoingMessageSender {
         Self {
             verification_auth: OnceLock::new(),
             verification_connections: Mutex::new(HashSet::new()),
+            human_approval_connections: Mutex::new(HashSet::new()),
             next_server_request_id: AtomicI64::new(0),
             process_reply_sender: sender.clone(),
             sender,
@@ -304,7 +307,18 @@ impl OutgoingMessageSender {
         }
     }
 
+    pub(crate) async fn enable_human_approval_connection(&self, connection_id: ConnectionId) {
+        self.human_approval_connections
+            .lock()
+            .await
+            .insert(connection_id);
+    }
+
     pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) {
+        self.human_approval_connections
+            .lock()
+            .await
+            .remove(&connection_id);
         self.disconnect_user_verification_connection(connection_id)
             .await;
         let mut request_contexts = self.request_contexts.lock().await;
@@ -387,6 +401,22 @@ impl OutgoingMessageSender {
                     || verification_identity != self.verification_identity())
         };
         let (tx_approve, rx_approve) = oneshot::channel();
+        let human_connections;
+        let connection_ids = if requires_human_approval(&request) {
+            let eligible = self.human_approval_connections.lock().await;
+            human_connections = connection_ids
+                .unwrap_or_default()
+                .iter()
+                .copied()
+                .filter(|id| eligible.contains(id))
+                .collect::<Vec<_>>();
+            if human_connections.is_empty() {
+                return (outgoing_message_id, rx_approve);
+            }
+            Some(human_connections.as_slice())
+        } else {
+            connection_ids
+        };
         // One app owns this ceremony. Reconnect and other subscribers cannot answer it.
         let verification_owner = if user_verification {
             let eligible = self.verification_connections.lock().await;
@@ -409,6 +439,11 @@ impl OutgoingMessageSender {
             request_id_to_callback.insert(
                 id,
                 PendingCallbackEntry {
+                    human_approval_recipients: if requires_human_approval(&request) {
+                        connection_ids.unwrap_or_default().iter().copied().collect()
+                    } else {
+                        HashSet::new()
+                    },
                     verification_owner,
                     verification_auth_revision,
                     verification_identity: verification_identity.clone(),
@@ -481,7 +516,24 @@ impl OutgoingMessageSender {
         thread_id: ThreadId,
     ) {
         let requests = self.pending_requests_for_thread(thread_id).await;
+        let human_approval = self
+            .human_approval_connections
+            .lock()
+            .await
+            .contains(&connection_id);
         for request in requests {
+            if requires_human_approval(&request) {
+                if !human_approval {
+                    continue;
+                }
+                // Register the replay recipient before sending so its response
+                // cannot race ahead of callback authorization.
+                let mut callbacks = self.request_id_to_callback.lock().await;
+                let Some(entry) = callbacks.get_mut(request.id()) else {
+                    continue;
+                };
+                entry.human_approval_recipients.insert(connection_id);
+            }
             if let Err(err) = self
                 .sender
                 .send(OutgoingEnvelope::ToConnection {
@@ -595,8 +647,18 @@ impl OutgoingMessageSender {
         connection_id: ConnectionId,
         id: &RequestId,
     ) -> Option<(RequestId, PendingCallbackEntry)> {
+        let human_approval = self
+            .human_approval_connections
+            .lock()
+            .await
+            .contains(&connection_id);
         let mut callbacks = self.request_id_to_callback.lock().await;
         let entry = callbacks.get(id)?;
+        if requires_human_approval(&entry.request)
+            && (!human_approval || !entry.human_approval_recipients.contains(&connection_id))
+        {
+            return None;
+        }
         if let Some(owner) = entry.verification_owner {
             if owner != connection_id {
                 return None;
@@ -952,6 +1014,11 @@ fn timestamped_server_notification(notification: ServerNotification) -> Outgoing
     })
 }
 
+fn requires_human_approval(request: &ServerRequest) -> bool {
+    matches!(request, ServerRequest::CommandExecutionRequestApproval { params, .. }
+        if params.kind == codex_app_server_protocol::CommandExecutionApprovalKind::HumanCommand)
+}
+
 #[cfg(test)]
 #[path = "thread_sample_cancel_context_tests.rs"]
 mod thread_sample_cancel_context_tests;

codex-rs/app-server/src/request_processors/initialize_processor.rs

@@ -141,6 +141,11 @@ impl InitializeRequestProcessor {
         {
             return Err(invalid_request("Already initialized"));
         }
+        if experimental_api_enabled && capabilities.required_human_approval {
+            self.outgoing
+                .enable_human_approval_connection(connection_id)
+                .await;
+        }
         if user_verification_enabled {
             self.outgoing
                 .enable_user_verification_connection(connection_id)