refactor(cli): move directory source policy into Rust

The decompression source-open callback previously both tested for a named
directory and returned the generic open failure. That left a source-resource
rejection in the C orchestration path even though Rust already owns the
per-file ordering.

Add a C-owned directory probe to the projection. Rust invokes it before source
opening for named inputs and rejects a positive result; the C callback retains
the private filesystem helper and exact directory diagnostic. Stdin bypasses
the probe as before, and optional codec callbacks plus the remaining source
open, asynchronous, and cleanup ordering are unchanged.

Test Plan:
- rustfmt +nightly --check --edition 2021 rust/src/fileio_asyncio.rs (passed)
- capped GCC syntax-only check of programs/fileio.c with all optional-format
  macros enabled (passed)
- capped Clang syntax-only check of programs/fileio.c with all optional-format
  macros enabled (passed)
- git diff --cached --check (passed)
- workspace cargo +nightly fmt --check was not clean because of an unrelated
  rust/src/zstdmt_compress.rs formatting diff; that file was left untouched
- no cargo build/test, make, fuzzers, or upstream tests were run
- commit signing was disabled because the configured GPG prompt hung

rust/src/fileio_asyncio.rs

@@ -671,6 +671,8 @@ const FIO_RUST_DECOMPRESS_SRC_UNKNOWN_SIZE: u64 = u64::MAX;
 
 pub type FIO_rust_decompress_src_open_fn =
     unsafe extern "C" fn(*mut c_void, *const c_char, *mut u64, *mut c_int) -> c_int;
+pub type FIO_rust_decompress_src_directory_fn =
+    unsafe extern "C" fn(*mut c_void, *const c_char) -> c_int;
 pub type FIO_rust_decompress_src_async_fn = unsafe extern "C" fn(*mut c_void, c_int);
 pub type FIO_rust_decompress_src_attach_fn = unsafe extern "C" fn(*mut c_void);
 pub type FIO_rust_decompress_src_close_fn = unsafe extern "C" fn(*mut c_void) -> c_int;
@@ -713,6 +715,7 @@ pub struct FIO_rust_decompress_file_projection_t {
     pub utime_destination: Option<FIO_rust_decompress_handler_fn>,
     pub remove_destination: Option<FIO_rust_decompress_dst_remove_fn>,
     pub remove_source_file: Option<FIO_rust_decompress_src_remove_fn>,
+    pub source_is_directory: Option<FIO_rust_decompress_src_directory_fn>,
 }
 
 const FIO_RUST_DECOMPRESS_FILE_CALLBACK_OFFSET: usize =
@@ -812,9 +815,13 @@ const _: () = {
             == FIO_RUST_DECOMPRESS_FILE_CALLBACK_OFFSET + 14 * size_of::<usize>()
     );
     assert!(
-        size_of::<FIO_rust_decompress_file_projection_t>()
+        std::mem::offset_of!(FIO_rust_decompress_file_projection_t, source_is_directory)
             == FIO_RUST_DECOMPRESS_FILE_CALLBACK_OFFSET + 15 * size_of::<usize>()
     );
+    assert!(
+        size_of::<FIO_rust_decompress_file_projection_t>()
+            == FIO_RUST_DECOMPRESS_FILE_CALLBACK_OFFSET + 16 * size_of::<usize>()
+    );
 };
 
 pub const FIO_RUST_ZSTD_OK: c_int = 0;
@@ -3073,6 +3080,15 @@ pub unsafe extern "C" fn FIO_rust_decompressFilename(
     assert!(!projection.dst_file_name.is_null());
     assert!(!projection.src_file_name.is_null());
 
+    if projection.source_is_stdin == 0 {
+        let source_is_directory = projection
+            .source_is_directory
+            .expect("source directory callback is required for named files");
+        if unsafe { source_is_directory(projection.opaque, projection.src_file_name) } != 0 {
+            return 1;
+        }
+    }
+
     let open_source = projection
         .open_source
         .expect("source open callback is required");
@@ -5869,6 +5885,8 @@ mod tests {
         destination_fds: Vec<c_int>,
         source_file_size: u64,
         source_is_regular: c_int,
+        source_is_directory: c_int,
+        source_directory_checks: usize,
         source_open_status: c_int,
         source_close_status: c_int,
         destination_open_status: c_int,
@@ -5892,6 +5910,15 @@ mod tests {
         state.source_open_status
     }
 
+    unsafe extern "C" fn decompress_policy_source_is_directory(
+        opaque: *mut c_void,
+        _src_file_name: *const c_char,
+    ) -> c_int {
+        let state = unsafe { &mut *opaque.cast::<DecompressPolicyState>() };
+        state.source_directory_checks += 1;
+        state.source_is_directory
+    }
+
     unsafe extern "C" fn decompress_policy_async(opaque: *mut c_void, async_mode: c_int) {
         let state = unsafe { &mut *opaque.cast::<DecompressPolicyState>() };
         state.events.push(DECOMPRESS_POLICY_ASYNC);
@@ -6023,9 +6050,23 @@ mod tests {
             utime_destination: Some(decompress_policy_utime_destination),
             remove_destination: Some(decompress_policy_remove_destination),
             remove_source_file: Some(decompress_policy_remove_source),
+            source_is_directory: Some(decompress_policy_source_is_directory),
         }
     }
 
+    #[test]
+    fn decompress_file_policy_rejects_directory_before_opening_source() {
+        let mut state = DecompressPolicyState {
+            source_is_directory: 1,
+            ..DecompressPolicyState::default()
+        };
+        let projection = decompress_policy_projection(&mut state, 0, 0, 0, 0, 0);
+
+        assert_eq!(unsafe { FIO_rust_decompressFilename(&projection) }, 1);
+        assert_eq!(state.source_directory_checks, 1);
+        assert!(state.events.is_empty());
+    }
+
     #[test]
     fn decompress_file_policy_orders_cleanup_and_successful_source_removal() {
         let mut state = DecompressPolicyState {