Emit exec-policy warnings for freshly loaded thread config (#31253)

## Why

App-server initialization captures exec-policy parse warnings only once.
Each `thread/start` then reloads config for that thread's cwd and
rereads its `.rules` files, so rules that become malformed after
initialization—or belong to a different project—are dropped without a
fresh warning to the requesting client.

## What changed

- Validate exec-policy rules against the freshly loaded per-thread
config during `thread/start`.
- Send the existing structured `configWarning` only to the connection
that requested the thread.
- Preserve nonfatal startup and avoid repeating an identical warning
already delivered during initialization.

This is intentionally separate from the TUI `legacy_core` cleanup in
#31179 and the managed-requirements fallback fix in #31188.

codex-rs/app-server/README.md

@@ -1301,7 +1301,7 @@ Event notifications are the server-initiated event stream for thread lifecycles,
 
 Thread realtime uses a separate thread-scoped notification surface. `thread/realtime/*` notifications are ephemeral transport events, not `ThreadItem`s, and are not returned by `thread/read`, `thread/resume`, or `thread/fork`.
 
-Recoverable configuration and initialization warnings use the existing `configWarning` notification: `{ summary, details?, path?, range? }`. App-server may emit it during initialization for config parsing and related setup diagnostics.
+Recoverable configuration and initialization warnings use the existing `configWarning` notification: `{ summary, details?, path?, range? }`. App-server may emit it during initialization for config parsing and related setup diagnostics, or to the requesting connection during `thread/start` when that thread's exec-policy rules fail to parse.
 
 Generic runtime warnings use the `warning` notification: `{ threadId?, message }`. App-server emits this for non-fatal warnings from the core event stream, including cases where not all enabled skills are included in the model-visible skills list for a session.
 

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

@@ -320,6 +320,16 @@ fn exec_policy_warning_location(err: &ExecPolicyError) -> (Option<String>, Optio
     }
 }
 
+fn exec_policy_config_warning(err: &ExecPolicyError) -> ConfigWarningNotification {
+    let (path, range) = exec_policy_warning_location(err);
+    ConfigWarningNotification {
+        summary: "Error parsing rules; custom rules not applied.".to_string(),
+        details: Some(err.to_string()),
+        path,
+        range,
+    }
+}
+
 fn app_text_range(range: &CoreTextRange) -> AppTextRange {
     AppTextRange {
         start: AppTextPosition {
@@ -612,14 +622,7 @@ pub async fn run_main_with_transport_options(
     }
 
     if let Ok(Some(err)) = check_execpolicy_for_warnings(&config.config_layer_stack).await {
-        let (path, range) = exec_policy_warning_location(&err);
-        let message = ConfigWarningNotification {
-            summary: "Error parsing rules; custom rules not applied.".to_string(),
-            details: Some(err.to_string()),
-            path,
-            range,
-        };
-        config_warnings.push(message);
+        config_warnings.push(exec_policy_config_warning(&err));
     }
 
     if let Some(warning) = project_config_warning(&config) {

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

@@ -447,7 +447,7 @@ impl MessageProcessor {
             outgoing.clone(),
             analytics_events_client.clone(),
             Arc::clone(&config),
-            config_warnings,
+            config_warnings.clone(),
             rpc_transport,
         );
         let marketplace_processor = MarketplaceRequestProcessor::new(
@@ -495,6 +495,7 @@ impl MessageProcessor {
             state_db.clone(),
             log_db,
             Arc::clone(&skills_watcher),
+            config_warnings,
         );
         let turn_processor = TurnRequestProcessor::new(
             auth_manager.clone(),

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

@@ -364,6 +364,7 @@ pub(crate) struct ThreadRequestProcessor {
     pub(super) log_db: Option<LogDbLayer>,
     pub(super) background_tasks: TaskTracker,
     pub(super) skills_watcher: Arc<SkillsWatcher>,
+    pub(super) initial_config_warnings: Arc<Vec<ConfigWarningNotification>>,
 }
 
 /// Outcome of trying to satisfy a resume request from an already loaded thread.
@@ -395,6 +396,7 @@ impl ThreadRequestProcessor {
         state_db: Option<StateDbHandle>,
         log_db: Option<LogDbLayer>,
         skills_watcher: Arc<SkillsWatcher>,
+        initial_config_warnings: Vec<ConfigWarningNotification>,
     ) -> Self {
         Self {
             auth_manager,
@@ -413,6 +415,7 @@ impl ThreadRequestProcessor {
             log_db,
             background_tasks: TaskTracker::new(),
             skills_watcher,
+            initial_config_warnings: Arc::new(initial_config_warnings),
         }
     }
 
@@ -968,6 +971,7 @@ impl ThreadRequestProcessor {
         };
         let request_trace = request_context.request_trace();
         let config_manager = self.config_manager.clone();
+        let initial_config_warnings = Arc::clone(&self.initial_config_warnings);
         let outgoing = Arc::clone(&listener_task_context.outgoing);
         let error_request_id = request_id.clone();
         let thread_start_task = async move {
@@ -990,6 +994,7 @@ impl ThreadRequestProcessor {
                 allow_provider_model_fallback,
                 experimental_raw_events,
                 request_trace,
+                initial_config_warnings,
             )
             .await
             {
@@ -1066,6 +1071,7 @@ impl ThreadRequestProcessor {
         allow_provider_model_fallback: bool,
         experimental_raw_events: bool,
         request_trace: Option<W3cTraceContext>,
+        initial_config_warnings: Arc<Vec<ConfigWarningNotification>>,
     ) -> Result<(), JSONRPCErrorError> {
         let thread_start_started_at = std::time::Instant::now();
         let requested_cwd = typesafe_overrides.cwd.clone();
@@ -1139,6 +1145,21 @@ impl ThreadRequestProcessor {
                 .map_err(|err| config_load_error(&err))?;
         }
 
+        if let Ok(Some(err)) =
+            codex_core::check_execpolicy_for_warnings(&config.config_layer_stack).await
+        {
+            let notification = crate::exec_policy_config_warning(&err);
+            if !initial_config_warnings.contains(&notification) {
+                listener_task_context
+                    .outgoing
+                    .send_server_notification_to_connections(
+                        &[request_id.connection_id],
+                        ServerNotification::ConfigWarning(notification),
+                    )
+                    .await;
+            }
+        }
+
         let environments = environments.unwrap_or_else(|| {
             listener_task_context
                 .thread_manager

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

@@ -7,6 +7,7 @@ use app_test_support::to_response;
 use base64::Engine;
 use base64::engine::general_purpose::URL_SAFE_NO_PAD;
 use codex_app_server_protocol::ClientInfo;
+use codex_app_server_protocol::ConfigWarningNotification;
 use codex_app_server_protocol::InitializeParams;
 use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCMessage;
@@ -18,6 +19,8 @@ use codex_app_server_protocol::ThreadLoadedListParams;
 use codex_app_server_protocol::ThreadLoadedListResponse;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
+use codex_core::config::set_project_trust_level;
+use codex_protocol::config_types::TrustLevel;
 use futures::SinkExt;
 use futures::StreamExt;
 use hmac::Hmac;
@@ -104,6 +107,106 @@ async fn websocket_transport_routes_per_connection_handshake_and_responses() ->
     Ok(())
 }
 
+#[tokio::test]
+async fn thread_start_routes_project_exec_policy_warning_to_requester() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), "never")?;
+
+    let project = TempDir::new()?;
+    std::fs::create_dir(project.path().join(".git"))?;
+    let rules_dir = project.path().join(".codex/rules");
+    std::fs::create_dir_all(&rules_dir)?;
+    let rules_path = rules_dir.join("broken.rules");
+    std::fs::write(&rules_path, "prefix_rule(")?;
+    set_project_trust_level(codex_home.path(), project.path(), TrustLevel::Trusted)?;
+
+    let (mut process, bind_addr) = spawn_websocket_server(codex_home.path()).await?;
+    let mut requester = connect_websocket(bind_addr).await?;
+    let mut other_client = connect_websocket(bind_addr).await?;
+
+    send_initialize_request(&mut requester, /*id*/ 1, "requester").await?;
+    read_response_for_id(&mut requester, /*id*/ 1).await?;
+    send_initialize_request(&mut other_client, /*id*/ 2, "other_client").await?;
+    read_response_for_id(&mut other_client, /*id*/ 2).await?;
+
+    send_request(
+        &mut requester,
+        "thread/start",
+        /*id*/ 3,
+        Some(serde_json::to_value(ThreadStartParams {
+            cwd: Some(project.path().display().to_string()),
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })?),
+    )
+    .await?;
+
+    let target_id = RequestId::Integer(3);
+    let warning_summary = "Error parsing rules; custom rules not applied.";
+    let is_exec_policy_warning = |notification: &JSONRPCNotification| {
+        notification.method == "configWarning"
+            && notification
+                .params
+                .as_ref()
+                .and_then(|params| params.get("summary"))
+                .and_then(serde_json::Value::as_str)
+                == Some(warning_summary)
+    };
+    let mut response = None;
+    let mut warning = None;
+    while response.is_none() || warning.is_none() {
+        match read_jsonrpc_message(&mut requester).await? {
+            JSONRPCMessage::Response(candidate) if candidate.id == target_id => {
+                response = Some(candidate);
+            }
+            JSONRPCMessage::Notification(candidate) if is_exec_policy_warning(&candidate) => {
+                warning = Some(candidate);
+            }
+            _ => {}
+        }
+    }
+
+    let _: ThreadStartResponse = to_response(response.context("missing thread/start response")?)?;
+    let warning: ConfigWarningNotification = serde_json::from_value(
+        warning
+            .context("missing exec-policy configWarning")?
+            .params
+            .context("configWarning should include params")?,
+    )?;
+    assert_eq!(
+        warning
+            .path
+            .as_deref()
+            .map(Path::new)
+            .and_then(Path::file_name),
+        Some(std::ffi::OsStr::new("broken.rules"))
+    );
+
+    match timeout(Duration::from_millis(250), async {
+        loop {
+            let message = read_jsonrpc_message(&mut other_client).await?;
+            if let JSONRPCMessage::Notification(notification) = message
+                && is_exec_policy_warning(&notification)
+            {
+                return Ok::<_, anyhow::Error>(notification);
+            }
+        }
+    })
+    .await
+    {
+        Ok(Ok(_)) => bail!("exec-policy configWarning leaked to another connection"),
+        Ok(Err(err)) => return Err(err),
+        Err(_) => {}
+    }
+
+    process
+        .kill()
+        .await
+        .context("failed to stop websocket app-server process")?;
+    Ok(())
+}
+
 #[tokio::test]
 async fn websocket_transport_serves_health_endpoints_on_same_listener() -> Result<()> {
     let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;

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

@@ -7,14 +7,18 @@ use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::to_response;
 use app_test_support::write_chatgpt_auth;
 use codex_app_server_protocol::AskForApproval;
+use codex_app_server_protocol::ConfigWarningNotification;
 use codex_app_server_protocol::JSONRPCError;
 use codex_app_server_protocol::JSONRPCMessage;
+use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::McpServerStartupState;
 use codex_app_server_protocol::McpServerStatusUpdatedNotification;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::SandboxMode;
 use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::TextPosition;
+use codex_app_server_protocol::TextRange;
 use codex_app_server_protocol::ThreadHistoryMode;
 use codex_app_server_protocol::ThreadSource;
 use codex_app_server_protocol::ThreadStartParams;
@@ -54,6 +58,17 @@ use super::analytics::wait_for_analytics_payload;
 
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
+const EXEC_POLICY_PARSE_WARNING_SUMMARY: &str = "Error parsing rules; custom rules not applied.";
+
+fn is_exec_policy_config_warning(notification: &JSONRPCNotification) -> bool {
+    notification.method == "configWarning"
+        && notification
+            .params
+            .as_ref()
+            .and_then(|params| params.get("summary"))
+            .and_then(Value::as_str)
+            == Some(EXEC_POLICY_PARSE_WARNING_SUMMARY)
+}
 
 async fn start_thread_with_model(
     mcp: &mut TestAppServer,
@@ -104,6 +119,124 @@ model = "gpt-5.4-mini"
     Ok(())
 }
 
+#[tokio::test]
+async fn thread_start_warns_for_exec_policy_parse_failure_after_initialize() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let rules_dir = codex_home.path().join("rules");
+    std::fs::create_dir_all(&rules_dir)?;
+    let rules_path = rules_dir.join("broken.rules");
+    std::fs::write(&rules_path, "prefix_rule(")?;
+    let rules_path = std::fs::canonicalize(rules_path)?;
+
+    let request_id = mcp
+        .send_thread_start_request_with_auto_env(ThreadStartParams::default())
+        .await?;
+    let response = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let _: ThreadStartResponse = to_response(response)?;
+
+    let notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_matching_notification(
+            "exec-policy configWarning",
+            is_exec_policy_config_warning,
+        ),
+    )
+    .await??;
+    let notification: ServerNotification = notification.try_into()?;
+    let ServerNotification::ConfigWarning(warning) = notification else {
+        anyhow::bail!("unexpected notification variant");
+    };
+    let ConfigWarningNotification {
+        summary,
+        details,
+        path,
+        range,
+    } = warning;
+    assert_eq!(
+        (summary, range),
+        (
+            "Error parsing rules; custom rules not applied.".to_string(),
+            Some(TextRange {
+                start: TextPosition {
+                    line: 1,
+                    column: 13,
+                },
+                end: TextPosition {
+                    line: 1,
+                    column: 13,
+                },
+            }),
+        )
+    );
+    let path = path.context("warning should include a path")?;
+    assert_eq!(
+        normalize_path_for_comparison(path),
+        normalize_path_for_comparison(&rules_path)
+    );
+    let details = details.context("warning should include details")?;
+    assert!(
+        details.contains("failed to parse rules file") && details.contains("broken.rules"),
+        "unexpected warning details: {details}"
+    );
+    assert!(
+        details.contains("Parse error"),
+        "unexpected warning details: {details}"
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn thread_start_does_not_repeat_initialize_exec_policy_warning() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let rules_dir = codex_home.path().join("rules");
+    std::fs::create_dir_all(&rules_dir)?;
+    std::fs::write(rules_dir.join("broken.rules"), "prefix_rule(")?;
+
+    let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_matching_notification(
+            "initialize exec-policy configWarning",
+            is_exec_policy_config_warning,
+        ),
+    )
+    .await??;
+
+    let request_id = mcp
+        .send_thread_start_request_with_auto_env(ThreadStartParams::default())
+        .await?;
+    let response = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let _: ThreadStartResponse = to_response(response)?;
+
+    let duplicate_warning = timeout(
+        std::time::Duration::from_millis(250),
+        mcp.read_stream_until_matching_notification(
+            "duplicate exec-policy configWarning",
+            is_exec_policy_config_warning,
+        ),
+    )
+    .await;
+    assert!(
+        duplicate_warning.is_err(),
+        "thread/start repeated the initialize exec-policy warning"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn thread_start_provider_model_fallback_uses_bedrock_static_catalog() -> Result<()> {
     let codex_home = TempDir::new()?;