Canonicalize default namespaces in gRPC subscription filters (#38650)

## What changed

- Normalize both tool invocations and subscription filters before matching them.
- Treat missing and empty namespaces as aliases for the `functions` namespace while preserving the namespace reported with each invocation.

## Testing

- Add a gRPC service test covering missing, empty, and explicit default namespaces on both sides of a filtered subscription.

GitOrigin-RevId: 409326f8ee6b3ab9a6ff5ceca0693558bc9c364d

codex-rs/code-mode-host/src/grpc/routing.rs

@@ -4,6 +4,7 @@ use std::sync::PoisonError;
 use codex_code_mode_protocol::CodeModeNestedToolCall;
 use codex_code_mode_protocol::grpc as proto;
 use codex_code_mode_protocol::host::MAX_FRAME_BYTES;
+use codex_protocol::ToolName;
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use prost::Message;
@@ -99,6 +100,9 @@ impl GrpcSession {
             name: invocation.tool_name.name,
             namespace: invocation.tool_name.namespace,
         };
+        let canonical_tool_name =
+            ToolName::new(tool_name.namespace.clone(), tool_name.name.clone())
+                .with_default_namespace();
         let (sequence, subscriptions) = {
             let state = self.state.lock().unwrap_or_else(PoisonError::into_inner);
             let Some(execution) = state.cells.get(&cell_id) else {
@@ -113,7 +117,9 @@ impl GrpcSession {
                 .filter(|subscription| {
                     subscription.filters.is_empty()
                         || subscription.filters.iter().any(|filter| {
-                            filter.name == tool_name.name && filter.namespace == tool_name.namespace
+                            ToolName::new(filter.namespace.clone(), filter.name.clone())
+                                .with_default_namespace()
+                                == canonical_tool_name
                         })
                 })
                 .map(|subscription| (subscription.id, subscription.sender.clone()))

codex-rs/code-mode-host/src/grpc/service_tests.rs

@@ -164,6 +164,80 @@ async fn execute_stream_starts_immediately_and_wait_preserves_missing_cells() {
     ));
 }
 
+/// Missing, empty, and explicit default namespaces identify the same subscribed tool.
+#[tokio::test]
+async fn filtered_subscriptions_match_default_namespace_aliases() {
+    let aliases = [
+        (None, Some("functions".to_string())),
+        (Some(String::new()), None),
+        (Some("functions".to_string()), Some(String::new())),
+    ];
+
+    for (tool_namespace, filter_namespace) in aliases {
+        let host = GrpcCodeModeHost::new();
+        let (session_id, _events) = open_session(&host).await;
+        let mut subscription = host
+            .subscribe_to_tool_calls(Request::new(proto::SubscribeToToolCallsRequest {
+                session_id: session_id.clone(),
+                tool_names: vec![proto::ToolName {
+                    name: "echo".to_string(),
+                    namespace: filter_namespace,
+                }],
+            }))
+            .await
+            .expect("subscribe with a default-namespace alias")
+            .into_inner();
+        let mut request = execute_request(
+            &session_id,
+            "default-namespace",
+            "text(await tools.echo({}));",
+        );
+        request.yield_time_ms = Some(/*value*/ 60_000);
+        let mut enabled_tool = tool("echo");
+        enabled_tool.tool_name = Some(proto::ToolName {
+            name: "echo".to_string(),
+            namespace: tool_namespace.clone(),
+        });
+        request.enabled_tools = vec![enabled_tool];
+        let (_cell_id, mut execution) = execute_events(&host, request).await;
+
+        let invocation = subscription
+            .next()
+            .await
+            .expect("matching default-namespace invocation")
+            .expect("tool invocation");
+        assert_eq!(
+            invocation.tool_name,
+            Some(proto::ToolName {
+                name: "echo".to_string(),
+                namespace: tool_namespace,
+            })
+        );
+        host.complete_tool_call(Request::new(proto::CompleteToolCallRequest {
+            session_id,
+            invocation_id: invocation.invocation_id,
+            outcome: Some(proto::complete_tool_call_request::Outcome::Succeeded(
+                proto::ToolCallSucceeded {
+                    output_json: br#""done""#.to_vec(),
+                },
+            )),
+        }))
+        .await
+        .expect("complete the matching tool call");
+        assert!(matches!(
+            execution.next().await,
+            Some(Ok(proto::ExecuteEvent {
+                event: Some(proto::execute_event::Event::Outcome(
+                    proto::ExecutionOutcome {
+                        outcome: Some(proto::execution_outcome::Outcome::Completed(_)),
+                        ..
+                    }
+                )),
+            }))
+        ));
+    }
+}
+
 #[tokio::test]
 async fn filtered_subscriptions_receive_ordered_calls_and_unary_completions() {
     let host = GrpcCodeModeHost::new();