diff --git a/crates/onestore-smb/README.md b/crates/onestore-smb/README.md
index e379d79821866a87061eb418e82151b006c42cf5..9228557fd90324876b5b1e0592dc318846cbb0ac 100644
--- a/crates/onestore-smb/README.md
+++ b/crates/onestore-smb/README.md
@@ -38,6 +38,21 @@ identity is checked after acquiring the guards. Connection loss retires the
 client. Reconnect for subsequent operations, and reconcile an `Unknown` edit
 before retrying it. The transport does not automatically replay requests.
 
+`Client::read_dir(path, entry_limit)` enumerates a directory, including the share
+root with an empty path. It follows every response page and returns no partial
+list on interruption, entry-limit overflow or close failure. Entries retain exact
+Unicode names, observed sizes and MS-FSCC attributes, including directory/reparse
+flags. Concurrent directory changes are not an atomic snapshot; repeated names
+are rejected with `ResourceBusy`. Notebook identities come from the files, not
+directory names or sizes. Missing paths, denied access and non-directory paths
+have distinct I/O error kinds.
+
+`python3 tools/test_smb_directory.py VM OUTPUT` checks a caller-owned disposable
+Linux lab VM against its filesystem listing and interrupts directory requests,
+responses and close. It creates synthetic files in that VM; the caller retains
+responsibility for VM teardown. The parser also has bounded-record/truncation
+tests independent of the server.
+
 Device and simulator builds link for iOS. Native acceptance uses disposable
 OneNote 2010 clients and Samba; it does not establish on-device execution or
 physical power-loss durability.
diff --git a/crates/onestore-smb/src/directory.rs b/crates/onestore-smb/src/directory.rs
new file mode 100644
index 0000000000000000000000000000000000000000..4997431a09200a47462e6573c515ead57de50eb9
--- /dev/null
+++ b/crates/onestore-smb/src/directory.rs
@@ -0,0 +1,219 @@
+use super::*;
+use smb2::msg::query_directory::{
+    FileInformationClass, QueryDirectoryFlags, QueryDirectoryRequest, QueryDirectoryResponse,
+};
+use std::collections::BTreeMap;
+
+/// Observed directory metadata, not a stable notebook identity or a file snapshot.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DirectoryEntry {
+    pub name: String,
+    pub size: u64,
+    /// MS-FSCC file attributes; directory is 0x10 and reparse point is 0x400.
+    pub attributes: u32,
+}
+
+impl Client {
+    /// Enumerates a share-relative directory completely or returns an error without a partial list.
+    /// An empty path selects the share root. The limit excludes `.` and `..`.
+    /// Concurrent directory changes are not an atomic snapshot; repeated names return ResourceBusy.
+    pub fn read_dir(&self, path: &str, limit: usize) -> io::Result<Vec<DirectoryEntry>> {
+        if path.contains('\0') || path.encode_utf16().count() > 32767 {
+            return Err(io::ErrorKind::InvalidInput.into());
+        }
+        let response: CreateResponse = self.request(
+            Command::Create,
+            CreateRequest {
+                requested_oplock_level: OplockLevel::None,
+                impersonation_level: ImpersonationLevel::Impersonation,
+                desired_access: FileAccessMask::new(0x80000000),
+                file_attributes: 0,
+                share_access: ShareAccess(7),
+                create_disposition: CreateDisposition::FileOpen,
+                create_options: 0x1,
+                name: smb2::encode_path(&path.replace('\\', "/")),
+                create_contexts: Vec::new(),
+            },
+        )?;
+        let file = File {
+            client: self,
+            id: Some(response.file_id),
+        };
+        let mut entries = BTreeMap::new();
+        loop {
+            let response: io::Result<QueryDirectoryResponse> =
+                self.request_with(Command::QueryDirectory, |connection| {
+                    (
+                        QueryDirectoryRequest {
+                            file_information_class: FileInformationClass::FileDirectoryInformation,
+                            flags: QueryDirectoryFlags(if entries.is_empty() { 1 } else { 0 }),
+                            file_index: 0,
+                            file_id: file.id.unwrap(),
+                            output_buffer_length: connection
+                                .params()
+                                .expect("connected SMB session is negotiated")
+                                .max_transact_size
+                                .min(65536),
+                            file_name: "*".into(),
+                        },
+                        CreditCharge(1),
+                    )
+                });
+            let response = match response {
+                Ok(response) => response,
+                Err(error)
+                    if matches!(
+                        error.get_ref().and_then(|error| error.downcast_ref::<smb2::Error>()),
+                        Some(smb2::Error::Protocol { status, command: Command::QueryDirectory })
+                            if status.0 == 0x80000006 || (entries.is_empty() && status.0 == 0xc000000f)
+                    ) =>
+                {
+                    break;
+                }
+                Err(error) => return Err(error),
+            };
+            for entry in decode(&response.output_buffer)? {
+                if entries
+                    .insert(entry.name, (entry.size, entry.attributes))
+                    .is_some()
+                {
+                    return Err(io::ErrorKind::ResourceBusy.into());
+                }
+                if entries.len()
+                    - usize::from(entries.contains_key("."))
+                    - usize::from(entries.contains_key(".."))
+                    > limit
+                {
+                    return Err(io::ErrorKind::FileTooLarge.into());
+                }
+            }
+        }
+        file.close()?;
+        Ok(entries
+            .into_iter()
+            .filter(|(name, _)| name != "." && name != "..")
+            .map(|(name, (size, attributes))| DirectoryEntry {
+                name,
+                size,
+                attributes,
+            })
+            .collect())
+    }
+}
+
+fn decode(mut bytes: &[u8]) -> io::Result<Vec<DirectoryEntry>> {
+    if bytes.len() > 65536 {
+        return Err(io::ErrorKind::InvalidData.into());
+    }
+    let mut entries = Vec::new();
+    loop {
+        if bytes.len() < 64 {
+            return Err(io::ErrorKind::InvalidData.into());
+        }
+        let next = u32::from_le_bytes(bytes[..4].try_into().unwrap()) as usize;
+        let length = u32::from_le_bytes(bytes[60..64].try_into().unwrap()) as usize;
+        let end = 64usize
+            .checked_add(length)
+            .ok_or(io::ErrorKind::InvalidData)?;
+        if length == 0
+            || !length.is_multiple_of(2)
+            || end > bytes.len()
+            || (next != 0 && (next < end || !next.is_multiple_of(8) || next >= bytes.len()))
+            || (next == 0 && bytes.len() - end > 7)
+        {
+            return Err(io::ErrorKind::InvalidData.into());
+        }
+        let units: Vec<_> = bytes[64..end]
+            .chunks_exact(2)
+            .map(|unit| u16::from_le_bytes([unit[0], unit[1]]))
+            .collect();
+        let name = String::from_utf16(&units).map_err(|_| io::ErrorKind::InvalidData)?;
+        if name.contains(['\0', '/', '\\']) {
+            return Err(io::ErrorKind::InvalidData.into());
+        }
+        let size = i64::from_le_bytes(bytes[40..48].try_into().unwrap());
+        entries.push(DirectoryEntry {
+            name,
+            size: size.try_into().map_err(|_| io::ErrorKind::InvalidData)?,
+            attributes: u32::from_le_bytes(bytes[56..60].try_into().unwrap()),
+        });
+        if next == 0 {
+            return Ok(entries);
+        }
+        bytes = &bytes[next..];
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn record(name: &str, size: u64, attributes: u32) -> Vec<u8> {
+        let mut bytes = vec![0; 64];
+        bytes[40..48].copy_from_slice(&size.to_le_bytes());
+        bytes[56..60].copy_from_slice(&attributes.to_le_bytes());
+        let name: Vec<_> = name.encode_utf16().flat_map(u16::to_le_bytes).collect();
+        bytes[60..64].copy_from_slice(&(name.len() as u32).to_le_bytes());
+        bytes.extend(name);
+        bytes
+    }
+
+    #[test]
+    fn directory_records_preserve_names_sizes_and_attributes() {
+        let mut bytes = Vec::new();
+        let mut expected = Vec::new();
+        for i in 0..300u32 {
+            let name = format!("Section {i} 🦀 e\u{301}.one");
+            let size = u64::from(i) * 100_000_000;
+            let attributes = if i % 3 == 0 { 0x410 } else { 0x20 };
+            let mut entry = record(&name, size, attributes);
+            if i != 299 {
+                entry.resize(entry.len().next_multiple_of(8), 0xa5);
+                let length = entry.len() as u32;
+                entry[..4].copy_from_slice(&length.to_le_bytes());
+            }
+            bytes.extend(entry);
+            expected.push(DirectoryEntry {
+                name,
+                size,
+                attributes,
+            });
+        }
+        assert_eq!(decode(&bytes).unwrap(), expected);
+        for end in 0..bytes.len() {
+            assert!(decode(&bytes[..end]).is_err(), "accepted prefix {end}");
+        }
+    }
+
+    #[test]
+    fn invalid_directory_records_never_become_partial_results() {
+        assert!(decode(&vec![0; 65537]).is_err());
+        for name in ["", "bad\0name", "a/b", "a\\b"] {
+            assert!(decode(&record(name, 0, 0)).is_err());
+        }
+        let valid = record("a.one", 12, 0x20);
+        for (at, value) in [
+            (0, 8),
+            (0, 65),
+            (0, 72),
+            (0, u32::MAX),
+            (60, 0),
+            (60, 1),
+            (60, u32::MAX),
+            (44, u32::MAX),
+        ] {
+            let mut bytes = valid.clone();
+            bytes[at..at + 4].copy_from_slice(&value.to_le_bytes());
+            assert!(decode(&bytes).is_err(), "offset={at} value={value}");
+        }
+        let mut bytes = valid.clone();
+        bytes[64..66].copy_from_slice(&0xd800u16.to_le_bytes());
+        assert!(decode(&bytes).is_err());
+        let mut bytes = valid;
+        bytes.extend_from_slice(&[0; 8]);
+        assert!(decode(&bytes).is_err());
+        for name in [".", ".."] {
+            assert_eq!(decode(&record(name, 0, 0x10)).unwrap()[0].name, name);
+        }
+    }
+}
diff --git a/crates/onestore-smb/src/lib.rs b/crates/onestore-smb/src/lib.rs
index f8bf1c9155a4c0c4fbea52a43b6ca5e6b4b1720e..4deb6eace482a2b6a2ef8f22f3f58f84ef4a9482 100644
--- a/crates/onestore-smb/src/lib.rs
+++ b/crates/onestore-smb/src/lib.rs
@@ -25,6 +25,9 @@ use smb2::{
 use std::{io, ops::Range, sync::Mutex, time::Duration};
 use tokio::runtime::{Handle, Runtime};
 
+mod directory;
+pub use directory::DirectoryEntry;
+
 #[derive(Default)]
 pub struct Credentials<'a> {
     pub username: &'a str,
@@ -153,6 +156,8 @@ impl Client {
                 0xc0000043 | 0xc0000054 | 0xc0000055 => io::ErrorKind::WouldBlock,
                 0xc0000011 => io::ErrorKind::UnexpectedEof,
                 0xc0000034 | 0xc000003a => io::ErrorKind::NotFound,
+                0xc0000022 => io::ErrorKind::PermissionDenied,
+                0xc0000103 => io::ErrorKind::NotADirectory,
                 _ => io::ErrorKind::Other,
             };
             return Err(io::Error::new(
diff --git a/crates/onestore-smb/src/tests.rs b/crates/onestore-smb/src/tests.rs
index fb20e9dfc177b2049b494c563f36f741b9d0cf8a..8080ccc50e8173026f79313896148925b46e814d 100644
--- a/crates/onestore-smb/src/tests.rs
+++ b/crates/onestore-smb/src/tests.rs
@@ -10,6 +10,87 @@ use std::{
 
 mod faults;
 
+#[test]
+#[ignore = "requires an owned Samba directory and ONESTORE_SMB_DIRECTORY_ORACLE from its local filesystem"]
+fn live_directory() {
+    let client = client();
+    let bytes = fs::read(std::env::var("ONESTORE_SMB_DIRECTORY_ORACLE").unwrap()).unwrap();
+    let oracle: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+    let root = oracle["path"].as_str().unwrap();
+    let expected = oracle["entries"].as_array().unwrap();
+    let entries = client.read_dir(root, expected.len()).unwrap();
+    assert_eq!(entries.len(), expected.len());
+    for expected in expected {
+        let entry = entries
+            .iter()
+            .find(|entry| entry.name == expected["name"])
+            .unwrap();
+        let directory = expected["directory"].as_bool().unwrap();
+        assert_eq!(entry.attributes & 0x10 != 0, directory, "{}", entry.name);
+        if !directory {
+            assert_eq!(
+                entry.size,
+                expected["size"].as_u64().unwrap(),
+                "{}",
+                entry.name
+            );
+        }
+    }
+    assert_eq!(
+        client.read_dir(root, entries.len() - 1).unwrap_err().kind(),
+        io::ErrorKind::FileTooLarge
+    );
+    assert_eq!(client.read_dir(root, entries.len()).unwrap(), entries);
+    assert!(
+        client
+            .read_dir(&format!("{root}/empty"), 0)
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(
+        client
+            .read_dir(&format!("{root}/missing"), 1)
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::NotFound
+    );
+    assert_eq!(
+        client
+            .read_dir(&format!("{root}/file.one"), 1)
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::NotADirectory
+    );
+    assert_eq!(
+        client
+            .read_dir(&format!("{root}/denied"), 1)
+            .unwrap_err()
+            .kind(),
+        io::ErrorKind::PermissionDenied
+    );
+    assert!(
+        client
+            .read_dir("", 10_000)
+            .unwrap()
+            .iter()
+            .any(|entry| entry.name == root)
+    );
+}
+
+#[test]
+#[ignore = "requires an owned Samba directory through a proxy that interrupts a later directory page or close"]
+fn live_directory_interruption() {
+    let client = client();
+    let root = std::env::var("ONESTORE_SMB_DIRECTORY").unwrap();
+    let started = Instant::now();
+    assert!(client.read_dir(&root, 10_000).is_err());
+    assert!(started.elapsed() < Duration::from_secs(10));
+    assert_eq!(
+        client.read_dir(&root, 10_000).unwrap_err().kind(),
+        io::ErrorKind::NotConnected
+    );
+}
+
 fn client() -> Client {
     Client::connect(
         &std::env::var("ONESTORE_SMB_LAB").unwrap(),
diff --git a/tools/test_smb_directory.py b/tools/test_smb_directory.py
new file mode 100644
index 0000000000000000000000000000000000000000..6b91864b0582385ff14b534e78253ad1a2ae5213
--- /dev/null
+++ b/tools/test_smb_directory.py
@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""Compare SMB enumeration with a disposable Linux VM's filesystem and interrupt pagination."""
+import argparse
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import sys
+import time
+import uuid
+
+sys.path.insert(0, str(Path(__file__).resolve().parent / 'w7'))
+from linux_vm import load_instance, ssh_argv
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('vm', help='An already running, caller-owned Linux lab VM')
+    parser.add_argument('output', type=Path)
+    args = parser.parse_args()
+    output = args.output.resolve()
+    output.mkdir(parents=True, exist_ok=False)
+    config = load_instance(args.vm)
+    root = 'directory-' + uuid.uuid4().hex[:12]
+    script = f'''from pathlib import Path
+import json
+root = Path('/srv/agent') / {root!r}
+root.mkdir()
+(root / 'empty').mkdir()
+(root / 'nested 🦀').mkdir()
+(root / 'denied').mkdir()
+(root / 'file.one').write_bytes(b'fixture')
+for i in range(4000):
+    name = f'Section {{i:04d}} 🦀 é ' + 'x' * 80 + '.one'
+    with (root / name).open('wb') as file: file.truncate(i * 12345)
+entries = [dict(name=p.name, directory=p.is_dir(), size=p.stat().st_size) for p in root.iterdir()]
+(root / 'denied').chmod(0)
+print(json.dumps(dict(path=root.name, entries=entries), ensure_ascii=False))
+'''
+    fixture = subprocess.run(ssh_argv(args.vm, 'python3 -'), input=script, text=True,
+                             capture_output=True, check=True, timeout=60)
+    (output / 'oracle.json').write_text(fixture.stdout)
+    cases = [('complete', {})]
+    cases += [(f'{direction}-{occurrence}', dict(cut=14, direction=direction, occurrence=occurrence))
+              for direction in ('request', 'response') for occurrence in (1, 2, 10)]
+    cases.append(('close', dict(cut=6, direction='response')))
+    for name, control in cases:
+        with socket.socket() as reservation:
+            reservation.bind(('127.0.0.1', 0))
+            port = reservation.getsockname()[1]
+        control_path = output / f'{name}-control.json'
+        control_path.write_text(json.dumps(control))
+        trace = output / f'{name}-trace.jsonl'
+        with trace.open('w') as log:
+            proxy = subprocess.Popen([sys.executable, str(Path(__file__).with_name('smb-proxy.py')),
+                                      str(control_path), '--port', str(port), '--server', '127.0.0.1',
+                                      '--server-port', str(config['samba_port'])], stdout=log, stderr=log)
+            try:
+                deadline = time.monotonic() + 5
+                while True:
+                    records = [json.loads(line) for line in trace.read_text().splitlines()]
+                    if any('listening' in row for row in records) and any('control' in row for row in records):
+                        break
+                    if proxy.poll() is not None or time.monotonic() > deadline:
+                        raise RuntimeError('Directory test proxy did not start')
+                    time.sleep(.05)
+                env = dict(os.environ, ONESTORE_SMB_LAB=f'127.0.0.1:{port}',
+                           ONESTORE_SMB_DIRECTORY=root, ONESTORE_SMB_DIRECTORY_ORACLE=str(output / 'oracle.json'))
+                test = 'tests::live_directory_interruption' if control else 'tests::live_directory'
+                with (output / f'{name}.log').open('w') as result:
+                    subprocess.run(['cargo', 'test', '-p', 'onestore-smb', test, '--', '--ignored', '--exact'],
+                                   env=env, stdout=result, stderr=result, check=True, timeout=120)
+            finally:
+                proxy.terminate()
+                proxy.wait(timeout=5)
+        records = [json.loads(line) for line in trace.read_text().splitlines()]
+        if control:
+            assert sum('cut' in row for row in records) == 1
+        else:
+            assert sum(row.get('command') == 14 and row.get('status') == '0x0' for row in records) > 10
+            assert any(row.get('command') == 14 and row.get('status') == '0x80000006' for row in records)
+        print(f'{name}: passed', flush=True)
+    env = dict(os.environ, ONESTORE_SMB_LAB=f'127.0.0.1:{config["samba_port"]}',
+               ONESTORE_SMB_DIRECTORY_ORACLE=str(output / 'oracle.json'))
+    with (output / 'reconnected.log').open('w') as result:
+        subprocess.run(['cargo', 'test', '-p', 'onestore-smb', 'tests::live_directory', '--', '--ignored', '--exact'],
+                       env=env, stdout=result, stderr=result, check=True, timeout=120)
+    print('reconnected: passed', flush=True)
+
+
+if __name__ == '__main__':
+    main()