feat: limit checkout indexing to entrances and expose bounded typed discovery

Content directories always have no indexable children; direct authorized paths remain resolvable. Add List using provider Read/enumeration authority and separate List permission, dirs-first lexical live continuation, bounded retention/path bytes, scoped result coordinates and External DTO validation. List/Glob/Grep return Descriptor-typed entries without eager acquisition. Bind content Interfaces coherently to the live entrance scope.

Validation: root cargo check; FS 58 unit+6 pagination, Workdir 125+5 (default and http-client), Tools 84 unit+39 integration+1 doctest, Worker 901 unit+110 integration, Server 691 unit+11 CLI, Server API 83 tests; exact crates.io WIP 0.2.0 resolution; format/diff checks. Focused model/local/remote/provider/race/scope checks passed. Commands and semantics in docs/development/wip-integration-validation.md.

Publication/review paused: continuation ShowTicket returned HTTP 401. No Backend bypass, source push, MR approval, target integration or live environment update is claimed.

crates/fs-operation/src/lib.rs

@@ -941,6 +941,7 @@ mod tests {
         let list_error = run_list(
             &root,
             ListRequest {
+                after: None,
                 path: FsPath::new("selected-dir").unwrap(),
                 limit: 10,
             },
@@ -1008,6 +1009,7 @@ mod tests {
         let listed = run_list(
             &root,
             ListRequest {
+                after: None,
                 path: FsPath::new("notes").unwrap(),
                 limit: 10,
             },

crates/fs-operation/src/local.rs

@@ -1,3 +1,5 @@
+use std::cmp::Ordering;
+use std::collections::BinaryHeap;
 use std::ffi::OsString;
 use std::fs;
 use std::io::{BufReader, Read};
@@ -7,9 +9,9 @@ use sha2::{Digest, Sha256};
 
 use crate::{
     AtomicWriteMode, BoundedReadLimits, ContentHash, EditRequest, EditResult, EntryKind,
-    FsAccessPolicy, FsError, FsPath, ListEntry, ListRequest, ListResult, ReadBytesRequest,
-    ReadBytesResult, ReadRequest, ReadResult, StatRequest, StatResult, WriteRequest, WriteResult,
-    direct_symlink,
+    FsAccessPolicy, FsError, FsPath, ListCursor, ListEntry, ListRequest, ListResult,
+    ReadBytesRequest, ReadBytesResult, ReadRequest, ReadResult, StatRequest, StatResult,
+    WriteRequest, WriteResult, direct_symlink,
 };
 
 const READ_BUFFER_BYTES: usize = 16 * 1024;
@@ -437,12 +439,55 @@ pub fn run_edit(
     })
 }
 
+fn list_order(
+    left_kind: EntryKind,
+    left: &FsPath,
+    right_kind: EntryKind,
+    right: &FsPath,
+) -> Ordering {
+    (left_kind != EntryKind::Directory)
+        .cmp(&(right_kind != EntryKind::Directory))
+        .then_with(|| left.as_str().cmp(right.as_str()))
+}
+
+struct RetainedListEntry(ListEntry);
+
+impl PartialEq for RetainedListEntry {
+    fn eq(&self, other: &Self) -> bool {
+        self.cmp(other) == Ordering::Equal
+    }
+}
+impl Eq for RetainedListEntry {}
+impl Ord for RetainedListEntry {
+    fn cmp(&self, other: &Self) -> Ordering {
+        list_order(self.0.kind, &self.0.path, other.0.kind, &other.0.path)
+    }
+}
+impl PartialOrd for RetainedListEntry {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+fn list_entry_bytes(entry: &ListEntry) -> usize {
+    entry.path.as_str().len().saturating_add(64)
+}
+
 pub fn run_list(
     root: &Path,
     request: ListRequest,
     access: &dyn FsAccessPolicy,
 ) -> Result<ListResult, FsError> {
     let logical = request.path;
+    if let Some(after) = &request.after {
+        if after.path.is_root()
+            || Path::new(after.path.as_str()).parent() != Some(Path::new(logical.as_str()))
+        {
+            return Err(FsError::InvalidArgument(
+                "List cursor must name a direct child of the listed directory".into(),
+            ));
+        }
+    }
     access
         .check_cancelled()
         .map_err(|error| map_io(&logical, error))?;
@@ -462,11 +507,15 @@ pub fn run_list(
             total_entries: 0,
             total_bytes: 0,
             truncated: false,
+            next_after: None,
         });
     }
-    let mut entries = Vec::new();
+    let mut entries: BinaryHeap<RetainedListEntry> = BinaryHeap::new();
+    let mut retained_limit = request.limit.saturating_add(1);
     let mut retained_path_bytes = 0_usize;
-    let mut provider_truncated = false;
+    let mut total_entries = 0usize;
+    let mut total_bytes = 0u64;
+    let mut remaining_entries = 0usize;
     let read_dir = access
         .open_read_dir(&logical_base, &path)
         .map_err(|error| map_io(&logical, error))?;
@@ -513,34 +562,77 @@ pub fn run_list(
         let Ok(result_path) = FsPath::new(relative) else {
             continue;
         };
-        let retained = result_path.as_str().len().saturating_add(64);
-        if retained_path_bytes.saturating_add(retained) > crate::MAX_RESULT_PATH_BYTES {
-            provider_truncated = true;
-            break;
+        total_entries += 1;
+        total_bytes = total_bytes.saturating_add(metadata.len());
+        if request.after.as_ref().is_some_and(|after| {
+            list_order(kind, &result_path, after.kind, &after.path) != Ordering::Greater
+        }) {
+            continue;
         }
-        retained_path_bytes = retained_path_bytes.saturating_add(retained);
-        entries.push(ListEntry {
+        remaining_entries += 1;
+        let candidate = RetainedListEntry(ListEntry {
             path: result_path,
             kind,
             size: metadata.len(),
         });
+        if entries.len() == retained_limit {
+            if candidate >= *entries.peek().expect("nonempty List retention") {
+                continue;
+            }
+            retained_path_bytes -= list_entry_bytes(&entries.pop().unwrap().0);
+        }
+        retained_path_bytes = retained_path_bytes.saturating_add(list_entry_bytes(&candidate.0));
+        entries.push(candidate);
+        // Evict greatest keys, never stop scanning: enumeration order is
+        // unspecified and a later entry may precede every retained key.
+        while retained_path_bytes > crate::MAX_RESULT_PATH_BYTES {
+            retained_path_bytes -= list_entry_bytes(&entries.pop().unwrap().0);
+            // Never grow cardinality again after a byte-bound eviction: a
+            // later short path must not leap over a discarded earlier key.
+            retained_limit = entries.len();
+        }
+        if retained_limit == 0 {
+            return Err(FsError::InvalidArgument(
+                "List entry exceeds provider response bound".into(),
+            ));
+        }
     }
-    entries.sort_by(|left, right| {
-        let left_dir = left.kind == EntryKind::Directory;
-        let right_dir = right.kind == EntryKind::Directory;
-        right_dir
-            .cmp(&left_dir)
-            .then_with(|| left.path.as_str().cmp(right.path.as_str()))
-    });
-    let total_entries = entries.len();
-    let total_bytes = entries.iter().map(|entry| entry.size).sum();
-    let truncated = provider_truncated || entries.len() > request.limit;
+    let mut entries: Vec<_> = entries
+        .into_sorted_vec()
+        .into_iter()
+        .map(|entry| entry.0)
+        .collect();
     entries.truncate(request.limit);
+    // Continuation duplicates the last path on the wire; include it in the
+    // aggregate provider response bound too.
+    let mut response_bytes: usize = entries.iter().map(list_entry_bytes).sum();
+    while remaining_entries > entries.len()
+        && entries.last().is_some_and(|last| {
+            response_bytes.saturating_add(list_entry_bytes(last)) > crate::MAX_RESULT_PATH_BYTES
+        })
+    {
+        response_bytes -= list_entry_bytes(&entries.pop().unwrap());
+    }
+    if request.limit > 0 && remaining_entries > 0 && entries.is_empty() {
+        return Err(FsError::InvalidArgument(
+            "List entry exceeds provider response bound".into(),
+        ));
+    }
+    let truncated = remaining_entries > entries.len();
+    let next_after = if truncated {
+        entries.last().map(|entry| ListCursor {
+            kind: entry.kind,
+            path: entry.path.clone(),
+        })
+    } else {
+        None
+    };
     Ok(ListResult {
         entries,
         total_entries,
         total_bytes,
         truncated,
+        next_after,
     })
 }
 

crates/fs-operation/src/operation.rs

@@ -278,6 +278,22 @@ pub struct ListRequest {
     pub path: FsPath,
     #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
     pub limit: usize,
+    /// Exclusive live continuation in the same coordinates as returned entries.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub after: Option<ListCursor>,
+}
+
+/// A List ordering key: directories first, then lexical path (all other kinds
+/// share one ordering group). The cursor need not still exist. Each page scans
+/// the live directory: insertions before the key are not revisited, insertions
+/// after it may appear, and removals disappear. Renames or kind changes can move
+/// an entry across the key and cause omission or repetition; no snapshot or
+/// cross-page consistency is promised. Scoped wrappers translate coordinates.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct ListCursor {
+    pub kind: EntryKind,
+    pub path: FsPath,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
@@ -293,6 +309,11 @@ pub struct ListEntry {
 #[serde(deny_unknown_fields)]
 pub struct ListResult {
     pub entries: Vec<ListEntry>,
+    /// Last returned key when more entries follow. A zero-limit request has no
+    /// continuation even when truncated. Totals describe all visible entries
+    /// in this scan, including those at or before `after`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_after: Option<ListCursor>,
     #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
     pub total_entries: usize,
     #[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))]

crates/fs-operation/tests/list_pagination.rs

@@ -0,0 +1,248 @@
+use fs_operation::{
+    EntryKind, FsAccessPolicy, FsError, FsPath, ListCursor, ListRequest, ListResult,
+    MAX_RESULT_PATH_BYTES, run_list,
+};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+struct Access {
+    root: PathBuf,
+    checks: AtomicUsize,
+    cancel_at: usize,
+    enumerate: bool,
+}
+impl Access {
+    fn new(root: &Path) -> Self {
+        Self {
+            root: root.to_path_buf(),
+            checks: AtomicUsize::new(0),
+            cancel_at: usize::MAX,
+            enumerate: true,
+        }
+    }
+}
+impl FsAccessPolicy for Access {
+    fn is_readable(&self, path: &Path) -> bool {
+        path.starts_with(&self.root) && path != self.root.join("denied")
+    }
+    fn is_writable(&self, _: &Path) -> bool {
+        false
+    }
+    fn can_enumerate_directory(&self, _: &Path, _: &Path) -> bool {
+        self.enumerate
+    }
+    fn check_cancelled(&self) -> std::io::Result<()> {
+        if self.checks.fetch_add(1, Ordering::Relaxed) >= self.cancel_at {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::Interrupted,
+                "cancelled",
+            ));
+        }
+        Ok(())
+    }
+}
+fn path(value: &str) -> FsPath {
+    FsPath::new(value).unwrap()
+}
+fn page(root: &Path, limit: usize, after: Option<ListCursor>) -> ListResult {
+    run_list(
+        root,
+        ListRequest {
+            path: FsPath::root(),
+            limit,
+            after,
+        },
+        &Access::new(root),
+    )
+    .unwrap()
+}
+fn names(result: &ListResult) -> Vec<&str> {
+    result
+        .entries
+        .iter()
+        .map(|entry| entry.path.as_str())
+        .collect()
+}
+
+#[test]
+fn list_pages_order_directories_first_then_paths_without_duplicates() {
+    let root = tempfile::tempdir().unwrap();
+    for name in ["z", "b", "a"] {
+        std::fs::write(root.path().join(name), b"xx").unwrap();
+    }
+    for name in ["z-dir", "a-dir"] {
+        std::fs::create_dir(root.path().join(name)).unwrap();
+    }
+    let full = page(root.path(), usize::MAX, None);
+    assert_eq!(names(&full), ["a-dir", "z-dir", "a", "b", "z"]);
+    let mut after = None;
+    let mut entries = Vec::new();
+    loop {
+        let result = page(root.path(), 2, after);
+        assert!(result.entries.len() <= 2);
+        assert_eq!(result.total_entries, 5);
+        assert_eq!(result.total_bytes, full.total_bytes);
+        assert_eq!(result.truncated, result.next_after.is_some());
+        after = result.next_after;
+        entries.extend(result.entries);
+        if after.is_none() {
+            break;
+        }
+    }
+    assert_eq!(entries, full.entries);
+    let zero = page(root.path(), 0, None);
+    assert!(zero.entries.is_empty());
+    assert!(zero.truncated);
+    assert!(zero.next_after.is_none());
+    assert_eq!(zero.total_entries, 5);
+    let exhausted = page(
+        root.path(),
+        2,
+        Some(ListCursor {
+            kind: EntryKind::File,
+            path: path("z"),
+        }),
+    );
+    assert!(exhausted.entries.is_empty());
+    assert!(!exhausted.truncated);
+    assert!(exhausted.next_after.is_none());
+    assert_eq!(exhausted.total_entries, 5);
+}
+
+#[test]
+fn list_live_cursor_survives_removal_and_only_observes_new_keys_after_it() {
+    let root = tempfile::tempdir().unwrap();
+    for name in ["b", "d"] {
+        std::fs::write(root.path().join(name), b"x").unwrap();
+    }
+    let first = page(root.path(), 1, None);
+    assert_eq!(names(&first), ["b"]);
+    std::fs::remove_file(root.path().join("b")).unwrap();
+    for name in ["a", "c"] {
+        std::fs::write(root.path().join(name), b"xx").unwrap();
+    }
+    // All directories sort before a file cursor, even when created later.
+    std::fs::create_dir(root.path().join("z-directory")).unwrap();
+    let second = page(root.path(), 10, first.next_after);
+    assert_eq!(names(&second), ["c", "d"]);
+    assert_eq!(second.total_entries, 4);
+    assert!(!second.truncated);
+    assert!(second.next_after.is_none());
+    std::fs::remove_file(root.path().join("d")).unwrap();
+    std::fs::rename(root.path().join("a"), root.path().join("e")).unwrap();
+    let moved = page(
+        root.path(),
+        10,
+        Some(ListCursor {
+            kind: EntryKind::File,
+            path: path("d"),
+        }),
+    );
+    assert_eq!(names(&moved), ["e"]);
+}
+
+#[test]
+fn list_path_byte_bound_returns_a_continuation_instead_of_losing_the_tail() {
+    let root = tempfile::tempdir().unwrap();
+    let long_count = MAX_RESULT_PATH_BYTES / (240 + 64) + 10;
+    // A shorter late key must not leap over a longer evicted earlier key.
+    let count = long_count + 50;
+    let expected: Vec<_> = (0..count)
+        .map(|n| format!("{n:05}{}", "x".repeat(if n < long_count { 235 } else { 1 })))
+        .collect();
+    for name in expected.iter().rev() {
+        std::fs::write(root.path().join(name), b"x").unwrap();
+    }
+    let first = page(root.path(), usize::MAX, None);
+    assert!(first.truncated);
+    assert_eq!(first.total_entries, count);
+    assert_eq!(first.total_bytes, count as u64);
+    let path_bytes: usize = first
+        .entries
+        .iter()
+        .map(|entry| entry.path.as_str().len() + 64)
+        .sum::<usize>()
+        + first.next_after.as_ref().unwrap().path.as_str().len()
+        + 64;
+    assert!(path_bytes <= MAX_RESULT_PATH_BYTES);
+    let next = page(root.path(), usize::MAX, first.next_after.clone());
+    assert!(!next.truncated);
+    let got: Vec<_> = names(&first).into_iter().chain(names(&next)).collect();
+    assert_eq!(got, expected.iter().map(String::as_str).collect::<Vec<_>>());
+    let small = page(root.path(), 1, None);
+    assert_eq!(names(&small), [expected[0].as_str()]);
+    assert_eq!(small.total_entries, count);
+}
+
+#[test]
+fn list_keeps_ignored_and_hidden_entries_but_respects_read_and_enumeration_authority() {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::write(root.path().join(".gitignore"), "ignored\n").unwrap();
+    for name in ["ignored", ".hidden", "denied"] {
+        std::fs::write(root.path().join(name), b"x").unwrap();
+    }
+    let first = page(root.path(), 2, None);
+    assert_eq!(names(&first), [".gitignore", ".hidden"]);
+    let second = page(root.path(), 2, first.next_after);
+    assert_eq!(names(&second), ["ignored"]);
+    assert_eq!(second.total_entries, 3);
+    let mut access = Access::new(root.path());
+    access.enumerate = false;
+    let result = run_list(
+        root.path(),
+        ListRequest {
+            path: FsPath::root(),
+            limit: 1,
+            after: None,
+        },
+        &access,
+    )
+    .unwrap();
+    assert!(result.entries.is_empty());
+    assert!(!result.truncated);
+    assert!(result.next_after.is_none());
+}
+
+#[test]
+fn list_cancellation_is_checked_even_after_the_page_is_full() {
+    let root = tempfile::tempdir().unwrap();
+    for name in ["a", "b", "c"] {
+        std::fs::write(root.path().join(name), b"x").unwrap();
+    }
+    let mut access = Access::new(root.path());
+    access.cancel_at = 3;
+    let error = run_list(
+        root.path(),
+        ListRequest {
+            path: FsPath::root(),
+            limit: 1,
+            after: None,
+        },
+        &access,
+    )
+    .unwrap_err();
+    assert!(
+        matches!(error, FsError::Io { source, .. } if source.kind() == std::io::ErrorKind::Interrupted)
+    );
+}
+
+#[test]
+fn list_rejects_cursor_outside_the_requested_directory() {
+    let root = tempfile::tempdir().unwrap();
+    for value in ["", "other/a"] {
+        let error = run_list(
+            root.path(),
+            ListRequest {
+                path: FsPath::root(),
+                limit: 1,
+                after: Some(ListCursor {
+                    kind: EntryKind::File,
+                    path: path(value),
+                }),
+            },
+            &Access::new(root.path()),
+        )
+        .unwrap_err();
+        assert!(matches!(error, FsError::InvalidArgument(_)));
+    }
+}

crates/tools/README.md

@@ -38,6 +38,7 @@ capabilities remain authoritative. `Create` uses the `Write` capability/identity
 | Edit | `old_string`, `new_string`, `replace_all?` | Existing file; prior read/hash and provider validator required |
 | Write | `content` | Existing file only; prior read/hash and provider validator required |
 | Create | `path`, `content` | Bound parent directory; relative destination strictly below it; create-new, missing parents allowed |
+| List | `limit?` (1..1000, default 100), `after?` (`{kind, path}`) | Bound directory; Read capability; all typed `EntryKind` cursor values accepted, path is a direct child in checkout-root-relative provider coordinates |
 | Glob | `pattern`, `path?` | Bound directory; optional path relative beneath it |
 | Grep | `pattern`, `path?`, `glob?`, `type?`, `case_insensitive?`, `-B?`, `-A?`, `-C?`, `multiline?`, `output_mode?`, `head_limit?`, `offset?` | Bound directory; same provider search/options/rendering as Tool Grep |
 
@@ -55,6 +56,22 @@ The constructor's empty scope layers/root output are refined by provider wrapper
 Tools does not synthesize authority or parse/rebase rendered filenames. Ordinary
 Tool dispatch remains `session.glob`/`session.grep`, with its existing error mapping.
 
+List dispatches one bounded `checkout_search` page without Bash or entry prefetch.
+`CheckoutToolOutput.listing` preserves the typed `ListResult`, including
+`next_after`, for Host projection into canonical Worldspace entries and cursors.
+Its normal Tool output is a summary only, and `paths` is empty; every other
+Operation returns `listing: None`. The Host decodes canonical cursor paths before
+calling Tools, mapping the public `{kind, entry}` cursor into `{kind, path}`.
+Both returned entries and the continuation must stay directly inside the bound
+directory. Cursor kinds retain all typed values (`directory`, `file`, `symlink`,
+`other`); ordering groups directories first and all non-directories together,
+then compares paths. A truncated page must supply `next_after` matching its last
+entry's ordering key; an untruncated page has no continuation. Special typed
+listing entries are path coordinates, not checkout observations: direct checkout
+observation remains provider File/Directory-only. Pagination is live, not a
+cross-page snapshot; the same before/after bound-directory validator checks
+used for search apply to List.
+
 `CheckoutToolOutput.paths` contains typed checkout-root-relative search paths
 (from `GlobResult.paths` / `GrepResult.paths`) and the created path for Create.
 Rendered Grep text is never parsed to construct links. The optional validator is

crates/tools/src/checkout.rs

@@ -18,6 +18,10 @@ use crate::{ToolsError, Tracker, file_target::FileTarget};
 pub struct CheckoutToolOutput {
     pub output: ToolOutput,
     pub paths: Vec<WorkdirPath>,
+    /// Typed directory page, including its provider-coordinate continuation.
+    /// Special entries are path coordinates, not checked checkout observations.
+    /// The Host owns projection into canonical Worldspace entries and cursors.
+    pub listing: Option<workdir::ListResult>,
     pub validator: Option<Vec<u8>>,
 }
 
@@ -47,6 +51,7 @@ pub async fn execute_checkout_tool(
         .unwrap_or_default();
     let (capability, allowed): (_, &[&str]) = match tool_name {
         "Read" => (WorkdirSessionCapability::Read, &file_fields),
+        "List" => (WorkdirSessionCapability::Read, &["limit", "after"]),
         "Edit" => (WorkdirSessionCapability::Edit, &file_fields),
         "Write" => (WorkdirSessionCapability::Write, &file_fields),
         "Create" => (WorkdirSessionCapability::Write, &["path", "content"]),
@@ -132,7 +137,7 @@ pub async fn execute_checkout_tool(
             crate::write::execute_write(target, tracker, params.content, Some(destination), ctx)
                 .await
         }
-        "Glob" | "Grep" => {
+        "List" | "Glob" | "Grep" => {
             let before = selected
                 .session
                 .checkout_observe(path.clone())
@@ -141,7 +146,10 @@ pub async fn execute_checkout_tool(
             if before.validator != validator {
                 return Err(stale_search());
             }
-            let mut result = if tool_name == "Glob" {
+            let mut result = if tool_name == "List" {
+                let params: ListParams = decode(Value::Object(arguments), tool_name)?;
+                execute_list(selected.session.clone(), path.clone(), params).await?
+            } else if tool_name == "Glob" {
                 let params: crate::glob::GlobParams = decode(Value::Object(arguments), tool_name)?;
                 check_pattern(&params.pattern)?;
                 let search_path =
@@ -191,6 +199,120 @@ pub async fn execute_checkout_tool(
     }
 }
 
+/// Native List has no path override. `after.path` is already decoded by the Host
+/// into checkout-root-relative provider coordinates, not directory-relative.
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ListParams {
+    #[serde(default = "default_list_limit")]
+    limit: usize,
+    #[serde(default)]
+    after: Option<workdir::ListCursor>,
+}
+
+fn default_list_limit() -> usize {
+    100
+}
+
+async fn execute_list(
+    session: workdir::WorkdirSessionHandle,
+    path: WorkdirPath,
+    params: ListParams,
+) -> Result<CheckoutToolOutput, ToolError> {
+    if !(1..=1000).contains(&params.limit) {
+        return Err(ToolError::InvalidArgument(
+            "List limit must be between 1 and 1000".into(),
+        ));
+    }
+    if let Some(after) = &params.after {
+        if !is_direct_child(&path, &after.path) {
+            return Err(ToolError::InvalidArgument(
+                "List after must name a direct child of the bound directory".into(),
+            ));
+        }
+    }
+    let result = session
+        .checkout_search(workdir::CheckoutSearchRequest::new(
+            workdir::CheckoutSearchOperation::List(workdir::ListRequest {
+                path: path.clone(),
+                limit: params.limit,
+                after: params.after,
+            }),
+        ))
+        .await
+        .map_err(crate::file_target::checked_error)?;
+    let listing = match result {
+        workdir::CheckoutSearchResult::List(listing) => listing,
+        _ => return Err(crate::file_target::readonly_result_error("List")),
+    };
+    // Fail closed before the Host can publish typed links to provider results.
+    if listing.entries.len() > params.limit
+        || listing
+            .entries
+            .iter()
+            .any(|entry| !is_direct_child(&path, &entry.path))
+        || listing
+            .next_after
+            .as_ref()
+            .is_some_and(|after| !is_direct_child(&path, &after.path))
+    {
+        return Err(crate::file_target::readonly_result_error(
+            "List page escaped bound directory or exceeded its limit",
+        ));
+    }
+    // The provider ordering key is directory/non-directory group plus path.
+    // Preserve the typed kind; special entries remain continuable without
+    // acquiring a File/Directory-only checkout observation for them.
+    let valid_continuation = match (&listing.next_after, listing.entries.last()) {
+        (Some(after), Some(last)) if listing.truncated => {
+            after.path == last.path
+                && (after.kind == workdir::EntryKind::Directory)
+                    == (last.kind == workdir::EntryKind::Directory)
+        }
+        (None, _) => !listing.truncated,
+        _ => false,
+    };
+    if !valid_continuation {
+        return Err(crate::file_target::readonly_result_error(
+            "List continuation does not match the truncated page's last ordering key",
+        ));
+    }
+    let summary = format!(
+        "Listed {} of {} entries in {}{}",
+        listing.entries.len(),
+        listing.total_entries,
+        path,
+        if listing.truncated {
+            " (truncated)"
+        } else {
+            ""
+        },
+    );
+    Ok(CheckoutToolOutput {
+        output: ToolOutput {
+            summary,
+            content: None,
+            attachments: Vec::new(),
+        },
+        paths: Vec::new(),
+        listing: Some(listing),
+        validator: None,
+    })
+}
+
+fn is_direct_child(base: &WorkdirPath, path: &WorkdirPath) -> bool {
+    let Ok(logical) = WorkdirPath::new(path.as_str()) else {
+        return false;
+    };
+    let parent = std::path::Path::new(logical.as_str()).parent();
+    let expected = if base == &WorkdirPath::root() {
+        std::path::Path::new("")
+    } else {
+        std::path::Path::new(base.as_str())
+    };
+    logical != WorkdirPath::root() && parent == Some(expected)
+}
+
 fn decode<T: DeserializeOwned>(arguments: Value, name: &str) -> Result<T, ToolError> {
     crate::error::decode_file_value(arguments, name)
 }

crates/tools/src/checkout_tests.rs

@@ -85,6 +85,326 @@ impl Fixture {
     }
 }
 
+#[derive(Debug)]
+struct ListProvider {
+    workdir: workdir::Workdir,
+    capabilities: workdir::WorkdirSessionCapabilities,
+    result: workdir::CheckoutSearchResult,
+    change_validator: bool,
+    expected_after: Option<workdir::ListCursor>,
+    searched: std::sync::atomic::AtomicBool,
+}
+
+// This boundary adapter exposes only checked metadata/search. Any file content
+// read, ordinary list fallback, or command execution is a test failure.
+#[async_trait::async_trait]
+impl workdir::WorkdirSession for ListProvider {
+    fn workdir(&self) -> &workdir::Workdir {
+        &self.workdir
+    }
+    fn capabilities(&self) -> workdir::WorkdirSessionCapabilities {
+        self.capabilities
+    }
+    async fn checkout_observe(
+        &self,
+        path: WorkdirPath,
+    ) -> Result<workdir::CheckoutObservation, workdir::WorkdirError> {
+        assert_eq!(
+            path.as_str(),
+            "src",
+            "List observes only its bound directory"
+        );
+        Ok(workdir::CheckoutObservation {
+            path,
+            kind: workdir::EntryKind::Directory,
+            size: 0,
+            validator: vec![u8::from(
+                self.change_validator && self.searched.load(std::sync::atomic::Ordering::SeqCst),
+            )],
+            capabilities: self.capabilities,
+        })
+    }
+    async fn checkout_search(
+        &self,
+        request: workdir::CheckoutSearchRequest,
+    ) -> Result<workdir::CheckoutSearchResult, workdir::WorkdirError> {
+        let workdir::CheckoutSearchOperation::List(list) = request.operation else {
+            panic!("native List must dispatch a typed List request");
+        };
+        assert_eq!(list.after, self.expected_after);
+        assert!(
+            !self
+                .searched
+                .swap(true, std::sync::atomic::Ordering::SeqCst),
+            "List must not prefetch another page"
+        );
+        Ok(self.result.clone())
+    }
+    async fn stat(
+        &self,
+        _: workdir::StatRequest,
+    ) -> Result<workdir::StatResult, workdir::WorkdirError> {
+        panic!("native List must not call stat");
+    }
+    async fn read(
+        &self,
+        _: workdir::ReadRequest,
+    ) -> Result<workdir::ReadResult, workdir::WorkdirError> {
+        panic!("native List must not read file content");
+    }
+    async fn write(
+        &self,
+        _: workdir::WriteRequest,
+    ) -> Result<workdir::WriteResult, workdir::WorkdirError> {
+        panic!("native List must not write");
+    }
+    async fn edit(
+        &self,
+        _: workdir::EditRequest,
+    ) -> Result<workdir::EditResult, workdir::WorkdirError> {
+        panic!("native List must not edit");
+    }
+    async fn list(
+        &self,
+        _: workdir::ListRequest,
+    ) -> Result<workdir::ListResult, workdir::WorkdirError> {
+        panic!("native List must use checkout_search, not ordinary list");
+    }
+    async fn glob(
+        &self,
+        _: workdir::GlobRequest,
+    ) -> Result<workdir::GlobResult, workdir::WorkdirError> {
+        panic!("native List must not glob");
+    }
+    async fn grep(
+        &self,
+        _: workdir::GrepRequest,
+    ) -> Result<workdir::GrepResult, workdir::WorkdirError> {
+        panic!("native List must not grep");
+    }
+    async fn start_command(
+        &self,
+        _: workdir::CommandRequest,
+    ) -> Result<workdir::CommandHandle, workdir::WorkdirError> {
+        panic!("native List must not execute commands");
+    }
+    async fn command_status(
+        &self,
+        _: workdir::CommandHandle,
+    ) -> Result<workdir::CommandStatus, workdir::WorkdirError> {
+        panic!("native List must not use commands");
+    }
+    async fn command_output(
+        &self,
+        _: workdir::CommandOutputRequest,
+    ) -> Result<workdir::CommandOutput, workdir::WorkdirError> {
+        panic!("native List must not use commands");
+    }
+    async fn cancel_command(&self, _: workdir::CommandHandle) -> Result<(), workdir::WorkdirError> {
+        panic!("native List must not use commands");
+    }
+    async fn close(&self) -> Result<(), workdir::WorkdirError> {
+        Ok(())
+    }
+}
+
+async fn list_with_provider(
+    capabilities: workdir::WorkdirSessionCapabilities,
+    result: workdir::CheckoutSearchResult,
+    change_validator: bool,
+) -> Result<CheckoutToolOutput, ToolError> {
+    list_with_provider_after(capabilities, result, change_validator, None).await
+}
+
+async fn list_with_provider_after(
+    capabilities: workdir::WorkdirSessionCapabilities,
+    result: workdir::CheckoutSearchResult,
+    change_validator: bool,
+    after: Option<workdir::ListCursor>,
+) -> Result<CheckoutToolOutput, ToolError> {
+    let router = Arc::new(WorkdirSessionRouter::new());
+    router
+        .attach(
+            WorkdirAttachmentAlias::new("selected").unwrap(),
+            Arc::new(ListProvider {
+                workdir: workdir::Workdir::new("list-test"),
+                capabilities,
+                result,
+                change_validator,
+                expected_after: after.clone(),
+                searched: std::sync::atomic::AtomicBool::new(false),
+            }),
+        )
+        .unwrap();
+    let generation = router.resolve(Some("selected")).unwrap().generation;
+    execute_checkout_tool(
+        router,
+        Tracker::new(),
+        "selected",
+        generation,
+        WorkdirPath::new("src").unwrap(),
+        vec![0],
+        "List",
+        json!({"limit": 1, "after": after}),
+        Default::default(),
+    )
+    .await
+}
+
+fn list_page() -> workdir::ListResult {
+    workdir::ListResult {
+        entries: vec![workdir::ListEntry {
+            path: WorkdirPath::new("src/a.txt").unwrap(),
+            kind: workdir::EntryKind::File,
+            size: 5,
+        }],
+        total_entries: 2,
+        total_bytes: 10,
+        truncated: true,
+        next_after: Some(fs_operation::ListCursor {
+            kind: workdir::EntryKind::File,
+            path: WorkdirPath::new("src/a.txt").unwrap(),
+        }),
+    }
+}
+
+#[tokio::test]
+async fn checkout_list_uses_read_capability_and_checked_search_without_prefetch() {
+    let read_only = workdir::WorkdirSessionCapabilities::EMPTY.with(WorkdirSessionCapability::Read);
+    let page = list_page();
+    let result = list_with_provider(
+        read_only,
+        workdir::CheckoutSearchResult::List(page.clone()),
+        false,
+    )
+    .await
+    .unwrap();
+    assert_eq!(result.listing, Some(page));
+    assert_eq!(result.validator, Some(vec![0]));
+    let no_read = workdir::WorkdirSessionCapabilities::EMPTY.with(WorkdirSessionCapability::Glob);
+    let error = list_with_provider(
+        no_read,
+        workdir::CheckoutSearchResult::List(list_page()),
+        false,
+    )
+    .await
+    .err()
+    .unwrap();
+    assert!(matches!(error, ToolError::InvalidArgument(_)));
+    assert!(error.to_string().contains("Read"));
+}
+
+#[tokio::test]
+async fn checkout_list_rejects_changed_post_search_directory_validator() {
+    let error = list_with_provider(
+        workdir::WorkdirSessionCapabilities::READ_ONLY,
+        workdir::CheckoutSearchResult::List(list_page()),
+        true,
+    )
+    .await
+    .err()
+    .unwrap();
+    assert_checkout_stale(error);
+}
+
+#[tokio::test]
+async fn checkout_list_rejects_provider_pages_that_escape_or_exceed_bounds() {
+    let mut outside = list_page();
+    outside.entries[0].path = WorkdirPath::new("outside.txt").unwrap();
+    let mut nested = list_page();
+    nested.entries[0].path = WorkdirPath::new("src/nested/a.txt").unwrap();
+    let mut cursor_outside = list_page();
+    cursor_outside.next_after.as_mut().unwrap().path = WorkdirPath::new("outside.txt").unwrap();
+    let mut cursor_kind = list_page();
+    cursor_kind.next_after.as_mut().unwrap().kind = workdir::EntryKind::Directory;
+    let mut oversized = list_page();
+    oversized.entries.push(oversized.entries[0].clone());
+    for result in [
+        workdir::CheckoutSearchResult::List(outside),
+        workdir::CheckoutSearchResult::List(nested),
+        workdir::CheckoutSearchResult::List(cursor_outside),
+        workdir::CheckoutSearchResult::List(cursor_kind),
+        workdir::CheckoutSearchResult::List(oversized),
+        workdir::CheckoutSearchResult::Glob(workdir::GlobResult {
+            paths: vec![],
+            truncated: false,
+        }),
+    ] {
+        let error = list_with_provider(
+            workdir::WorkdirSessionCapabilities::READ_ONLY,
+            result,
+            false,
+        )
+        .await
+        .err()
+        .unwrap();
+        assert_checkout_code(error, "checkout_unavailable");
+    }
+}
+
+#[tokio::test]
+async fn checkout_list_special_entry_cursors_preserve_kind_and_continue_the_page() {
+    // Provider-only special entries stay path coordinates: the adapter forbids
+    // content reads and does not offer direct observations for these entries.
+    for kind in [workdir::EntryKind::Other, workdir::EntryKind::Symlink] {
+        let mut first_page = list_page();
+        first_page.entries[0].kind = kind;
+        first_page.next_after.as_mut().unwrap().kind = kind;
+        let first = list_with_provider(
+            workdir::WorkdirSessionCapabilities::READ_ONLY,
+            workdir::CheckoutSearchResult::List(first_page),
+            false,
+        )
+        .await
+        .unwrap()
+        .listing
+        .unwrap();
+        assert_eq!(first.entries[0].kind, kind);
+        let after = first.next_after.unwrap();
+        assert_eq!(after.kind, kind);
+        let mut second_page = list_page();
+        second_page.entries[0].path = WorkdirPath::new("src/z.txt").unwrap();
+        second_page.truncated = false;
+        second_page.next_after = None;
+        let second = list_with_provider_after(
+            workdir::WorkdirSessionCapabilities::READ_ONLY,
+            workdir::CheckoutSearchResult::List(second_page.clone()),
+            false,
+            Some(after),
+        )
+        .await
+        .unwrap()
+        .listing
+        .unwrap();
+        assert_eq!(second, second_page);
+    }
+}
+
+#[tokio::test]
+async fn checkout_list_rejects_missing_or_mismatched_truncated_continuations() {
+    let mut missing = list_page();
+    missing.next_after = None;
+    let mut empty = list_page();
+    empty.entries.clear();
+    let mut wrong_path = list_page();
+    wrong_path.next_after.as_mut().unwrap().path = WorkdirPath::new("src/z.txt").unwrap();
+    let mut wrong_group = list_page();
+    wrong_group.next_after.as_mut().unwrap().kind = workdir::EntryKind::Directory;
+    let mut untruncated = list_page();
+    untruncated.truncated = false;
+    for page in [missing, empty, wrong_path, wrong_group, untruncated] {
+        let error = list_with_provider(
+            workdir::WorkdirSessionCapabilities::READ_ONLY,
+            workdir::CheckoutSearchResult::List(page),
+            false,
+        )
+        .await
+        .err()
+        .unwrap();
+        assert_checkout_code(error, "checkout_unavailable");
+    }
+}
+
 fn assert_checkout_stale(error: ToolError) {
     assert_checkout_code(error, "checkout_stale");
 }
@@ -557,6 +877,183 @@ async fn checkout_search_typed_paths_and_normal_rendering() {
         let normal = fixture.normal("Grep", json!({"target_workdir": "main", "path": "src", "pattern": "hit", "output_mode": mode, "-C": 1})).await.unwrap();
         assert_eq!(grep.output.summary, normal.summary);
         assert_eq!(grep.output.content, normal.content);
+        assert!(grep.listing.is_none());
         assert!(!grep.output.content.unwrap().contains("outside.txt"));
     }
 }
+
+#[tokio::test]
+async fn checkout_list_returns_one_typed_page_and_provider_coordinate_continuation() {
+    let fixture = Fixture::new();
+    std::fs::create_dir_all(fixture.dir.path().join("src/nested")).unwrap();
+    std::fs::write(fixture.dir.path().join("src/a.txt"), "alpha").unwrap();
+    std::fs::write(fixture.dir.path().join("src/z.txt"), "z").unwrap();
+    std::fs::write(fixture.dir.path().join("outside.txt"), "outside").unwrap();
+    let (_, validator) = fixture.observation("main", "src").await;
+    let first = fixture
+        .native("main", "src", "List", json!({"limit": 1}))
+        .await
+        .unwrap();
+    assert!(first.paths.is_empty());
+    assert!(first.output.content.is_none());
+    assert!(first.output.summary.contains("Listed 1 of 3 entries"));
+    assert!(first.output.summary.contains("truncated"));
+    assert_eq!(first.validator, Some(validator));
+    let page = first.listing.unwrap();
+    assert_eq!(page.entries.len(), 1);
+    assert_eq!(page.entries[0].path.as_str(), "src/nested");
+    assert_eq!(page.entries[0].kind, workdir::EntryKind::Directory);
+    assert!(page.truncated);
+    let total_bytes = page.total_bytes;
+    assert!(total_bytes >= 6);
+    let cursor = page.next_after.unwrap();
+    assert_eq!(cursor.path.as_str(), "src/nested");
+    let second = fixture
+        .native("main", "src", "List", json!({"after": cursor}))
+        .await
+        .unwrap();
+    let page = second.listing.unwrap();
+    assert_eq!(
+        page.entries
+            .iter()
+            .map(|entry| entry.path.as_str())
+            .collect::<Vec<_>>(),
+        ["src/a.txt", "src/z.txt"]
+    );
+    assert_eq!(page.total_entries, 3);
+    assert_eq!(page.total_bytes, total_bytes);
+    assert!(!page.truncated);
+    assert!(page.next_after.is_none());
+    assert_eq!(fixture.tracker.change_stat(), crate::ChangeStat::default());
+}
+
+#[tokio::test]
+async fn checkout_list_root_file_cursor_stays_on_selected_attachment() {
+    let fixture = Fixture::new();
+    std::fs::write(fixture.dir.path().join("a.txt"), "a").unwrap();
+    std::fs::write(fixture.dir.path().join("z.txt"), "z").unwrap();
+    let other = TempDir::new().unwrap();
+    std::fs::write(other.path().join("wrong.txt"), "wrong").unwrap();
+    fixture
+        .router
+        .attach(
+            WorkdirAttachmentAlias::new("other").unwrap(),
+            Arc::new(LocalWorkdirSession::new(
+                Scope::writable(other.path()).unwrap(),
+                other.path().to_path_buf(),
+            )),
+        )
+        .unwrap();
+    let first = fixture
+        .native("main", ".", "List", json!({"limit": 1}))
+        .await
+        .unwrap()
+        .listing
+        .unwrap();
+    let cursor = first.next_after.unwrap();
+    assert_eq!(cursor.kind, workdir::EntryKind::File);
+    assert_eq!(cursor.path.as_str(), "a.txt");
+    let second = fixture
+        .native("main", ".", "List", json!({"after": cursor}))
+        .await
+        .unwrap()
+        .listing
+        .unwrap();
+    assert_eq!(second.entries.len(), 1);
+    assert_eq!(second.entries[0].path.as_str(), "z.txt");
+    let other_page = fixture
+        .native("other", ".", "List", json!({}))
+        .await
+        .unwrap()
+        .listing
+        .unwrap();
+    assert_eq!(other_page.entries.len(), 1);
+    assert_eq!(other_page.entries[0].path.as_str(), "wrong.txt");
+}
+
+#[tokio::test]
+async fn checkout_list_defaults_to_100_and_accepts_both_limit_bounds() {
+    let fixture = Fixture::new();
+    for index in 0..101 {
+        std::fs::write(fixture.dir.path().join(format!("{index:03}.txt")), "").unwrap();
+    }
+    for (arguments, count, truncated) in [
+        (json!({}), 100, true),
+        (json!({"limit": 1}), 1, true),
+        (json!({"limit": 1000}), 101, false),
+    ] {
+        let result = fixture
+            .native("main", ".", "List", arguments)
+            .await
+            .unwrap();
+        let page = result.listing.unwrap();
+        assert_eq!(page.entries.len(), count);
+        assert_eq!(page.truncated, truncated);
+        assert_eq!(page.next_after.is_some(), truncated);
+    }
+}
+
+#[tokio::test]
+async fn checkout_list_rejects_route_overrides_malformed_limits_and_nonchild_cursors() {
+    let fixture = Fixture::new();
+    std::fs::create_dir(fixture.dir.path().join("src")).unwrap();
+    for arguments in [
+        json!({"path": "."}),
+        json!({"target_workdir": "other"}),
+        json!({"generation": 1}),
+        json!({"validator": []}),
+        json!({"limit": 0}),
+        json!({"limit": 1001}),
+        json!({"limit": -1}),
+        json!({"limit": 1.5}),
+        json!({"limit": "1"}),
+        json!({"limit": null}),
+        json!({"after": "src/a.txt"}),
+        json!({"after": {"path": "src/a.txt"}}),
+        json!({"after": {"kind": "file", "path": "src/a.txt", "extra": 1}}),
+        json!({"after": {"kind": "bogus", "path": "src/a.txt"}}),
+        json!({"after": {"kind": "file", "path": "src"}}),
+        json!({"after": {"kind": "directory", "path": "src/nested/deep"}}),
+        json!({"after": {"kind": "file", "path": "src-other/a.txt"}}),
+        json!({"after": {"kind": "file", "path": "a.txt"}}),
+        json!({"after": {"kind": "file", "path": "/src/a.txt"}}),
+        json!({"after": {"kind": "file", "path": "src/../a.txt"}}),
+        json!({"after": {"kind": "file", "path": "."}}),
+    ] {
+        let error = fixture
+            .native("main", "src", "List", arguments.clone())
+            .await
+            .err()
+            .unwrap();
+        assert!(
+            matches!(error, ToolError::InvalidArgument(_)),
+            "{arguments}: {error}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn checkout_list_requires_current_generation_and_directory_validator() {
+    let fixture = Fixture::new();
+    let (generation, validator) = fixture.observation("main", ".").await;
+    // No timing dependency: make the supplied observation explicitly stale.
+    let mut stale_validator = validator.clone();
+    stale_validator.push(0);
+    for (generation, validator) in [(generation, stale_validator), (generation + 1, validator)] {
+        let error = execute_checkout_tool(
+            fixture.router.clone(),
+            fixture.tracker.clone(),
+            "main",
+            generation,
+            WorkdirPath::root(),
+            validator,
+            "List",
+            json!({}),
+            Default::default(),
+        )
+        .await
+        .err()
+        .unwrap();
+        assert_checkout_stale(error);
+    }
+}

crates/tools/src/edit.rs

@@ -110,6 +110,7 @@ pub(crate) async fn execute_edit(
             attachments: Vec::new(),
         },
         paths: Vec::new(),
+        listing: None,
         validator,
     })
 }

crates/tools/src/glob.rs

@@ -93,6 +93,7 @@ pub(crate) async fn execute_glob(
             attachments: Vec::new(),
         },
         paths: result.paths,
+        listing: None,
         validator: None,
     })
 }

crates/tools/src/grep.rs

@@ -147,6 +147,7 @@ pub(crate) async fn execute_grep(
             attachments: Vec::new(),
         },
         paths: result.paths,
+        listing: None,
         validator: None,
     })
 }

crates/tools/src/read.rs

@@ -108,6 +108,7 @@ pub(crate) async fn execute_read(
             attachments: Vec::new(),
         },
         paths: Vec::new(),
+        listing: None,
         validator,
     })
 }

crates/tools/src/write.rs

@@ -125,6 +125,7 @@ pub(crate) async fn execute_write(
             attachments: Vec::new(),
         },
         paths: create_path.into_iter().collect(),
+        listing: None,
         validator,
     })
 }

crates/workdir/CHECKOUT_PROVIDER.md

@@ -101,6 +101,37 @@ Host generation reuse, native permission identity, read-history updates, model
 Operation publication, transport hosting (Runtime/Backend/External), and durable
 Ticket/MR review remain the parent integration's responsibilities.
 
+## Bounded live List pagination
+
+`ListRequest { path, limit, after: Option<ListCursor> }` accepts an exclusive
+ordering key `ListCursor { kind: EntryKind, path: FsPath }`. Directories sort
+first, then paths sort lexically within the directory/non-directory groups;
+File, Symlink and Other share the non-directory group. `ListResult.next_after`
+is the last returned key when another page remains. Reuse that key unchanged
+with the same directory and output root. Cursor paths must name a direct child
+in the result coordinate frame; they are not statted and need not still exist.
+The optional fields default to absent and are omitted from JSON when absent.
+
+The shared provider engine retains at most `limit + 1` candidates, additionally
+bounded by the 1 MiB aggregate path-response ceiling (including continuation),
+while scanning the directory once. It never stops merely because the page is
+full: enumeration order cannot determine the retained prefix. `total_entries`
+and `total_bytes` describe all visible entries in the current scan, including
+keys at/before `after`, not only the page or remaining suffix. A byte-bounded
+page may contain fewer than `limit` entries and still provide continuation.
+A zero-limit request returns totals and truncation but no continuation.
+List uses Read capability and read/enumeration scope, and **does not ignore-filter**
+`.gitignore`/`.ignore` entries or hide dotfiles; Glob/Grep keep their own policy.
+Cancellation and the 100,000-entry traversal ceiling apply to the complete scan,
+including excluded entries. External DTOs retain their existing item/path bounds
+and also validate cursor paths and continuation response bytes.
+
+Each page observes a live directory, not a snapshot, and no cross-page
+consistency is promised. Insertions after the cursor may appear; earlier keys
+are not revisited. Removed entries disappear, including the cursor entry.
+Renames or kind changes can move an entry across the cursor and cause omission
+or repetition. Even one scan is not transactional against external editors.
+
 ## Scoped checkout search boundary
 
 `CheckoutSearchRequest::new(CheckoutSearchOperation::{List, Glob, Grep}(request))`
@@ -110,7 +141,8 @@ are `scope_layers: Vec<Vec<WorkdirToolScopeRule>>` and `output_root: WorkdirPath
 Rules union within each layer; layers intersect with each other, the current
 provider authority, and the output-root boundary. Wrappers contribute their own
 scope; caller-supplied layers can only narrow, never replace it. Each wrapper
-translates all incoming coordinates once and forwards the composed output root.
+translates incoming provider coordinates once and forwards the composed output root.
+List cursors remain output-root-relative, just like result paths.
 Provider-rendered Grep text and every typed result path are already relative to
 that final root; wrappers must not prefix cwd onto returned paths again.
 

crates/workdir/src/checkout.rs

@@ -167,6 +167,19 @@ impl CheckoutSearchRequest {
         }
         let valid = match &self.operation {
             CheckoutSearchOperation::List(r) => {
+                if let Some(after) = &r.after {
+                    validate_path(&after.path)?;
+                    if after.path.is_root()
+                        || std::path::Path::new(after.path.as_str()).parent()
+                            != std::path::Path::new(r.path.as_str())
+                                .strip_prefix(self.output_root.as_str())
+                                .ok()
+                    {
+                        return Err(WorkdirError::InvalidArgument(
+                            "List cursor must name a direct child of the listed directory".into(),
+                        ));
+                    }
+                }
                 r.limit <= crate::external::MAX_EXTERNAL_RESULT_ITEMS
             }
             CheckoutSearchOperation::Glob(r) => {
@@ -793,6 +806,7 @@ mod tests {
         for provider in [&local as &dyn WorkdirSession, &external] {
             let listed = provider
                 .list(ListRequest {
+                    after: None,
                     path: WorkdirPath::root(),
                     limit: 100,
                 })

crates/workdir/src/checkout_search_tests.rs

@@ -60,6 +60,7 @@ fn fixture() -> tempfile::TempDir {
 async fn verify_sparse(session: &dyn WorkdirSession) {
     let listed = session
         .list(ListRequest {
+            after: None,
             path: WorkdirPath::root(),
             limit: 100,
         })
@@ -75,6 +76,7 @@ async fn verify_sparse(session: &dyn WorkdirSession) {
     );
     let empty = session
         .list(ListRequest {
+            after: None,
             path: path("denied"),
             limit: 100,
         })
@@ -108,6 +110,7 @@ async fn verify_sparse(session: &dyn WorkdirSession) {
     let CheckoutSearchResult::List(native) = session
         .checkout_search(CheckoutSearchRequest::new(CheckoutSearchOperation::List(
             ListRequest {
+                after: None,
                 path: WorkdirPath::root(),
                 limit: 100,
             },
@@ -161,6 +164,7 @@ async fn checkout_search_nested_sparse_nonrecursive_scopes_rebase_and_exclude_co
             .tool_session()
             .checkout_search(CheckoutSearchRequest::new(CheckoutSearchOperation::List(
                 ListRequest {
+                    after: None,
                     path: WorkdirPath::root(),
                     limit: 100
                 }
@@ -196,6 +200,7 @@ async fn checkout_search_composes_actual_nested_wrappers_and_output_root() {
     let CheckoutSearchResult::List(result) = stacked
         .checkout_search(CheckoutSearchRequest::new(CheckoutSearchOperation::List(
             ListRequest {
+                after: None,
                 path: path("keep"),
                 limit: 100,
             },
@@ -229,6 +234,7 @@ async fn checkout_search_composes_actual_nested_wrappers_and_output_root() {
     let entries = second
         .tool_session()
         .list(ListRequest {
+            after: None,
             path: WorkdirPath::root(),
             limit: 100,
         })
@@ -246,6 +252,7 @@ async fn checkout_search_composes_actual_nested_wrappers_and_output_root() {
         second
             .tool_session()
             .list(ListRequest {
+                after: None,
                 path: path("deep"),
                 limit: 100
             })
@@ -292,6 +299,7 @@ async fn checkout_search_scope_layers_intersect_before_list_and_grep() {
         root.path().to_path_buf(),
     );
     let mut request = CheckoutSearchRequest::new(CheckoutSearchOperation::List(ListRequest {
+        after: None,
         path: path("child"),
         limit: 100,
     }));
@@ -325,6 +333,7 @@ async fn checkout_search_wire_dto_bounds_and_provider_capabilities() {
         root.path().to_path_buf(),
     );
     let request = CheckoutSearchRequest::new(CheckoutSearchOperation::List(ListRequest {
+        after: None,
         path: path("child/keep"),
         limit: 100,
     }));
@@ -795,6 +804,7 @@ async fn checkout_search_preserves_current_provider_scope_and_rejects_symlink_es
     let CheckoutSearchResult::List(result) = source
         .checkout_search(CheckoutSearchRequest {
             operation: CheckoutSearchOperation::List(ListRequest {
+                after: None,
                 path: path("child/denied"),
                 limit: 100,
             }),

crates/workdir/src/external.rs

@@ -155,7 +155,11 @@ fn validate_external_operation(operation: &WorkdirSessionOperation) -> Result<()
                     <= MAX_EXTERNAL_SCOPE_RULES
                 && validate_external_operation(&match &request.operation {
                     crate::CheckoutSearchOperation::List(r) => {
-                        WorkdirSessionOperation::List(r.clone())
+                        // request.validate() checks the output_root-relative
+                        // cursor; validate the provider-coordinate path here.
+                        let mut list = r.clone();
+                        list.after = None;
+                        WorkdirSessionOperation::List(list)
                     }
                     crate::CheckoutSearchOperation::Glob(r) => {
                         WorkdirSessionOperation::Glob(r.clone())
@@ -212,7 +216,14 @@ fn validate_external_operation(operation: &WorkdirSessionOperation) -> Result<()
                     .contains(&request.max_bytes)
         }
         WorkdirSessionOperation::List(request) => {
-            is_root_relative(&request.path) && request.limit <= MAX_EXTERNAL_RESULT_ITEMS
+            is_root_relative(&request.path)
+                && request.limit <= MAX_EXTERNAL_RESULT_ITEMS
+                && request.after.as_ref().is_none_or(|after| {
+                    is_root_relative(&after.path)
+                        && !after.path.is_root()
+                        && Path::new(after.path.as_str()).parent()
+                            == Some(Path::new(request.path.as_str()))
+                })
         }
         WorkdirSessionOperation::Glob(request) => {
             is_root_relative(&request.path)
@@ -395,7 +406,20 @@ fn validate_external_result(result: &WorkdirSessionOperationResult) -> Result<()
                     .entries
                     .iter()
                     .all(|entry| is_root_relative(&entry.path))
-                && result_paths_fit(result.entries.iter().map(|entry| &entry.path))
+                && result.next_after.as_ref().is_none_or(|after| {
+                    is_root_relative(&after.path)
+                        && result.truncated
+                        && result.entries.last().is_some_and(|entry| {
+                            entry.kind == after.kind && entry.path == after.path
+                        })
+                })
+                && result_paths_fit(
+                    result
+                        .entries
+                        .iter()
+                        .map(|entry| &entry.path)
+                        .chain(result.next_after.iter().map(|after| &after.path)),
+                )
         }
         WorkdirSessionOperationResult::Glob(result) => {
             result.paths.len() <= MAX_EXTERNAL_RESULT_ITEMS
@@ -917,6 +941,7 @@ mod tests {
         assert!(serde_json::from_value::<ExternalWorkdirServerFrame>(nested).is_err());
 
         let unbounded = WorkdirSessionOperation::List(ListRequest {
+            after: None,
             path: WorkdirPath::root(),
             limit: MAX_EXTERNAL_RESULT_ITEMS + 1,
         });

crates/workdir/src/lib.rs

@@ -33,9 +33,9 @@ use tokio::sync::broadcast;
 
 pub use fs_operation::{
     BoundedReadLimits, ContentHash, EditRequest, EditResult, EntryKind, FsPath as WorkdirPath,
-    GlobRequest, GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest,
-    ListResult, ReadBytesRequest, ReadBytesResult, ReadRequest, ReadResult, StatRequest,
-    StatResult, WriteRequest, WriteResult,
+    GlobRequest, GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListCursor, ListEntry,
+    ListRequest, ListResult, ReadBytesRequest, ReadBytesResult, ReadRequest, ReadResult,
+    StatRequest, StatResult, WriteRequest, WriteResult,
 };
 pub use http::dispatch_workdir_session_operation;
 pub use local::{

crates/workdir/src/local.rs

@@ -2695,6 +2695,7 @@ mod tests {
         let listed = WorkdirSession::list(
             &workdir,
             ListRequest {
+                after: None,
                 path: WorkdirPath::new("notes").unwrap(),
                 limit: 10,
             },
@@ -3385,6 +3386,7 @@ mod tests {
         let list = WorkdirSession::list(
             &workdir,
             ListRequest {
+                after: None,
                 path: WorkdirPath::new("yoi.local").unwrap(),
                 limit: 10,
             },

crates/workdir/src/scope.rs

@@ -1005,6 +1005,9 @@ impl WorkdirSession for ScopedWorkdirSession {
             .await?;
         self.ensure_read(&path, request.operation.capability())?;
         *request.operation.path_mut() = path;
+        // List cursors stay output_root-relative, just like source results.
+        // Composing output_root below handles their coordinates without stat
+        // of a key that may have disappeared between live pages.
         request.output_root = self.resolve_path(&request.output_root)?;
         for layer in &mut request.scope_layers {
             for rule in layer {
@@ -1873,6 +1876,7 @@ mod tests {
         assert_eq!(
             parent
                 .list(ListRequest {
+                    after: None,
                     path: fs_path("child"),
                     limit: 100
                 })
@@ -1890,6 +1894,7 @@ mod tests {
             child
                 .tool_session()
                 .list(ListRequest {
+                    after: None,
                     path: FsPath::root(),
                     limit: 100
                 })
@@ -1901,6 +1906,7 @@ mod tests {
                 .tool_session()
                 .checkout_search(crate::CheckoutSearchRequest::new(
                     crate::CheckoutSearchOperation::List(ListRequest {
+                        after: None,
                         path: FsPath::root(),
                         limit: 100
                     })

crates/workdir/tests/list_pagination.rs

@@ -0,0 +1,267 @@
+#![cfg(target_os = "linux")]
+use std::sync::Arc;
+use workdir::external::{
+    ExternalWorkdirOperation, ExternalWorkdirOperationResult, MAX_EXTERNAL_RESULT_ITEMS,
+};
+use workdir::http::{WorkdirSessionOperation as Op, WorkdirSessionOperationResult as Output};
+use workdir::*;
+
+fn path(value: &str) -> WorkdirPath {
+    WorkdirPath::new(value).unwrap()
+}
+fn rule(value: &str, recursive: bool) -> WorkdirToolScopeRule {
+    WorkdirToolScopeRule {
+        target: path(value),
+        recursive,
+        permission: WorkdirToolScopePermission::Read,
+        symlink_policy: manifest::SymlinkPolicy::Logical,
+    }
+}
+fn scope(rules: Vec<WorkdirToolScopeRule>, cwd: &str) -> WorkdirToolScope {
+    WorkdirToolScope {
+        rules,
+        cwd: path(cwd),
+        command: false,
+    }
+}
+fn request(value: &str, limit: usize, after: Option<ListCursor>) -> ListRequest {
+    ListRequest {
+        path: path(value),
+        limit,
+        after,
+    }
+}
+fn names(result: &ListResult) -> Vec<&str> {
+    result
+        .entries
+        .iter()
+        .map(|entry| entry.path.as_str())
+        .collect()
+}
+
+#[tokio::test]
+async fn list_scoped_external_pages_rebase_cursor_and_preserve_sparse_authority() {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir_all(root.path().join("child/deep")).unwrap();
+    for name in ["child/a", "child/z", "child/deep/secret", "outside"] {
+        std::fs::write(root.path().join(name), b"x").unwrap();
+    }
+    let source = LocalWorkdirSession::external_read_only(
+        Workdir::new("pagination"),
+        root.path(),
+        BoundedReadLimits::EXTERNAL_DEFAULT,
+    )
+    .unwrap();
+    let broker = WorkdirToolBroker::new(Arc::new(source));
+    let lease = broker
+        .scope(scope(vec![rule("child", false)], "child"))
+        .await
+        .unwrap();
+    // Two actual wrappers must not duplicate child/ in the continuation.
+    let stacked = WorkdirToolBroker::new(lease.tool_session());
+    let first = stacked.list(request("", 1, None)).await.unwrap();
+    assert_eq!(names(&first), ["deep"]);
+    assert_eq!(first.next_after.as_ref().unwrap().path, path("deep"));
+    assert_eq!(
+        first.next_after.as_ref().unwrap().kind,
+        EntryKind::Directory
+    );
+    assert_eq!(first.total_entries, 3);
+    assert!(
+        stacked
+            .list(request("deep", 1, None))
+            .await
+            .unwrap()
+            .entries
+            .is_empty()
+    );
+    let second = stacked
+        .list(request("", 1, first.next_after))
+        .await
+        .unwrap();
+    assert_eq!(names(&second), ["a"]);
+    std::fs::remove_file(root.path().join("child/a")).unwrap();
+    std::fs::write(root.path().join("child/b"), b"xx").unwrap();
+    let third = stacked
+        .list(request("", 10, second.next_after))
+        .await
+        .unwrap();
+    assert_eq!(names(&third), ["b", "z"]);
+    assert!(!third.truncated);
+    assert!(third.next_after.is_none());
+    assert_eq!(third.total_entries, 3);
+    lease.close().await.unwrap();
+    assert!(matches!(
+        stacked.list(request("", 1, None)).await,
+        Err(WorkdirError::SessionClosed)
+    ));
+}
+
+#[tokio::test]
+async fn list_checkout_search_cursor_uses_output_root_coordinates_across_nested_wrappers() {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir_all(root.path().join("child/keep")).unwrap();
+    for name in ["a", "b", "c"] {
+        std::fs::write(root.path().join("child/keep").join(name), b"x").unwrap();
+    }
+    let source = LocalWorkdirSession::external_read_only(
+        Workdir::new("rebased-pagination"),
+        root.path(),
+        BoundedReadLimits::EXTERNAL_DEFAULT,
+    )
+    .unwrap();
+    let broker = WorkdirToolBroker::new(Arc::new(source));
+    let lease = broker
+        .scope(scope(vec![rule("child", true)], "child"))
+        .await
+        .unwrap();
+    let stacked = WorkdirToolBroker::new(lease.tool_session());
+    let mut search =
+        CheckoutSearchRequest::new(CheckoutSearchOperation::List(request("keep", 1, None)));
+    search.output_root = path("keep");
+    let CheckoutSearchResult::List(first) = stacked.checkout_search(search.clone()).await.unwrap()
+    else {
+        panic!()
+    };
+    assert_eq!(names(&first), ["a"]);
+    if let CheckoutSearchOperation::List(list) = &mut search.operation {
+        list.after = first.next_after;
+    }
+    let CheckoutSearchResult::List(second) = stacked.checkout_search(search).await.unwrap() else {
+        panic!()
+    };
+    assert_eq!(names(&second), ["b"]);
+    assert_eq!(second.next_after.unwrap().path, path("b"));
+}
+
+#[tokio::test]
+async fn list_read_capability_suffices_without_glob_and_missing_read_is_denied() {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::write(root.path().join("a"), b"x").unwrap();
+    let source = LocalWorkdirSession::materialized(
+        root.path().to_path_buf(),
+        root.path().to_path_buf(),
+        manifest::SharedScope::new(manifest::Scope::writable(root.path()).unwrap()),
+        WorkdirSessionCapabilities::from_capabilities([WorkdirSessionCapability::Read]),
+    );
+    let broker = WorkdirToolBroker::new(Arc::new(source));
+    let lease = broker.scope(scope(vec![rule("", true)], "")).await.unwrap();
+    assert_eq!(
+        names(
+            &lease
+                .tool_session()
+                .list(request("", 1, None))
+                .await
+                .unwrap()
+        ),
+        ["a"]
+    );
+    let denied = LocalWorkdirSession::external_with_capabilities_pinned(
+        Workdir::new("no-read"),
+        ExternalWorkdirRoot::pin(root.path()).unwrap(),
+        BoundedReadLimits::EXTERNAL_DEFAULT,
+        WorkdirSessionCapabilities::COMMAND_ONLY,
+    )
+    .unwrap();
+    assert!(matches!(
+        denied.list(request("", 1, None)).await,
+        Err(WorkdirError::Unsupported(WorkdirSessionCapability::Read))
+    ));
+}
+
+#[tokio::test]
+async fn list_external_wire_preserves_continuation_and_rejects_unbounded_cursor() {
+    let root = tempfile::tempdir().unwrap();
+    std::fs::create_dir(root.path().join("child")).unwrap();
+    for name in ["a", "b"] {
+        std::fs::write(root.path().join("child").join(name), b"x").unwrap();
+    }
+    let source = LocalWorkdirSession::external_read_only(
+        Workdir::new("wire-list"),
+        root.path(),
+        BoundedReadLimits::EXTERNAL_DEFAULT,
+    )
+    .unwrap();
+    let first = source.list(request("child", 1, None)).await.unwrap();
+    let wire = serde_json::to_value(
+        ExternalWorkdirOperationResult::try_from(Output::List(first.clone())).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(wire["result"]["next_after"]["path"], "child/a");
+    let decoded: ExternalWorkdirOperationResult = serde_json::from_value(wire.clone()).unwrap();
+    assert_eq!(decoded.into_inner(), Output::List(first.clone()));
+    let operation =
+        ExternalWorkdirOperation::try_from(Op::List(request("child", 1, first.next_after)))
+            .unwrap();
+    let decoded: ExternalWorkdirOperation =
+        serde_json::from_value(serde_json::to_value(operation).unwrap()).unwrap();
+    let Output::List(next) = dispatch_workdir_session_operation(&source, decoded.into_inner())
+        .await
+        .unwrap()
+    else {
+        panic!()
+    };
+    assert_eq!(names(&next), ["child/b"]);
+    assert!(next.next_after.is_none());
+    for bad_path in ["/host".to_owned(), "x".repeat(4097), "outside/a".to_owned()] {
+        let mut bad = serde_json::to_value(Op::List(request("child", 1, None))).unwrap();
+        bad["request"]["after"] = serde_json::json!({"kind": "file", "path": bad_path});
+        assert!(serde_json::from_value::<ExternalWorkdirOperation>(bad).is_err());
+        let mut bad_result = wire.clone();
+        bad_result["result"]["next_after"]["path"] = serde_json::json!(bad_path);
+        assert!(serde_json::from_value::<ExternalWorkdirOperationResult>(bad_result).is_err());
+    }
+    assert!(
+        ExternalWorkdirOperation::try_from(Op::List(request(
+            "child",
+            MAX_EXTERNAL_RESULT_ITEMS + 1,
+            None
+        )))
+        .is_err()
+    );
+    // Search cursors use result coordinates, not provider path coordinates.
+    let mut search = CheckoutSearchRequest::new(CheckoutSearchOperation::List(request(
+        "child",
+        1,
+        Some(ListCursor {
+            kind: EntryKind::File,
+            path: path("a"),
+        }),
+    )));
+    search.output_root = path("child");
+    let bounded = ExternalWorkdirOperation::try_from(Op::CheckoutSearch(search)).unwrap();
+    let Output::CheckoutSearch(CheckoutSearchResult::List(next)) =
+        dispatch_workdir_session_operation(&source, bounded.into_inner())
+            .await
+            .unwrap()
+    else {
+        panic!()
+    };
+    assert_eq!(names(&next), ["b"]);
+}
+
+#[test]
+fn list_external_result_counts_cursor_toward_aggregate_path_bound() {
+    let entries: Vec<_> = (0..252)
+        .map(|n| ListEntry {
+            path: path(&format!("{n:04}{}", "x".repeat(4092))),
+            kind: EntryKind::File,
+            size: 1,
+        })
+        .collect();
+    let after = ListCursor {
+        kind: EntryKind::File,
+        path: entries.last().unwrap().path.clone(),
+    };
+    let result = ListResult {
+        entries,
+        total_entries: 253,
+        total_bytes: 253,
+        truncated: true,
+        next_after: None,
+    };
+    let mut wire = serde_json::to_value(Output::List(result)).unwrap();
+    assert!(serde_json::from_value::<ExternalWorkdirOperationResult>(wire.clone()).is_ok());
+    wire["result"]["next_after"] = serde_json::to_value(after).unwrap();
+    assert!(serde_json::from_value::<ExternalWorkdirOperationResult>(wire).is_err());
+}

crates/worker/src/checkout.rs

@@ -11,11 +11,12 @@ use manifest::{ToolPermissionAction, ToolPermissionConfig};
 use serde_json::{Value as Json, json};
 use sha2::{Digest, Sha256};
 use wip_protocol::{
-    Documentation, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object, OperationDeclaration,
-    ParameterDeclaration, ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
+    Documentation, FieldDeclaration, INTERFACE_FORMAT_V1, InterfaceDescriptor, Object,
+    OperationDeclaration, ParameterDeclaration, ProtocolError, ProtocolErrorCode,
+    ReturnDeclaration, TypeExpr, Value,
 };
 use workdir::{
-    CheckoutObservation, EntryKind, ListRequest, WorkdirPath, WorkdirSessionCapability as Cap,
+    CheckoutObservation, EntryKind, WorkdirPath, WorkdirSessionCapability as Cap,
     WorkdirSessionRouter,
 };
 
@@ -126,11 +127,15 @@ impl Provider {
         }
         self.project(
             ROOT,
-            descriptor(vec![operation(
-                "list",
-                "List currently accessible checkouts",
-                vec![],
-            )]),
+            descriptor(vec![{
+                let mut op = operation(
+                    "list",
+                    "List accessible checkout entrances (not filesystem contents)",
+                    vec![],
+                );
+                op.returns.r#type = TypeExpr::Json;
+                op
+            }]),
             state.finalize().to_vec(),
             "collection",
             Arc::new(Collection {
@@ -169,8 +174,14 @@ impl Provider {
         }
     }
     async fn node(self: &Arc<Self>, path: &str) -> Result<Option<WipProjection>, ProtocolError> {
+        Ok(self.publication_inner(path).await?.map(|p| p.projection))
+    }
+    async fn publication_inner(
+        self: &Arc<Self>,
+        path: &str,
+    ) -> Result<Option<WipPublication>, ProtocolError> {
         if path == ROOT {
-            return Ok(Some(self.collection()));
+            return WipPublication::self_scoped(self.collection()).map(Some);
         }
         let Some((alias, target)) = self.parse(path) else {
             return Ok(None);
@@ -202,39 +213,86 @@ impl Provider {
         if observation.kind == EntryKind::File && !self.allowed("Read", &permission_input) {
             return Ok(None);
         }
+        let workdir = selected.session.workdir().id().as_str().to_owned();
+        let root = checkout_root(&alias);
+        // The router guard pins this connection across both observations. Root
+        // lifetime is attachment lifetime, not directory-content validator state.
+        let scope = if target.is_root() {
+            None
+        } else {
+            let root_observation =
+                match selected.session.checkout_observe(WorkdirPath::root()).await {
+                    Ok(o) => o,
+                    Err(_) => return Ok(None),
+                };
+            self.entry_projection(
+                &root,
+                &alias,
+                selected.generation,
+                &workdir,
+                WorkdirPath::root(),
+                root_observation,
+            )
+        };
+        let Some(projection) = self.entry_projection(
+            path,
+            &alias,
+            selected.generation,
+            &workdir,
+            target,
+            observation,
+        ) else {
+            return Ok(None);
+        };
+        if path == root {
+            return WipPublication::self_scoped(projection).map(Some);
+        }
+        let Some(scope) = scope else {
+            return Ok(None);
+        };
+        Ok(Some(WipPublication {
+            projection,
+            scope: Some(scope.object),
+        }))
+    }
+    fn entry_projection(
+        self: &Arc<Self>,
+        path: &str,
+        alias: &str,
+        generation: u64,
+        workdir: &str,
+        target: WorkdirPath,
+        observation: CheckoutObservation,
+    ) -> Option<WipProjection> {
         let ops = self.operations(&observation);
         if ops.is_empty() {
-            return Ok(None);
+            return None;
         }
         let kind = if observation.kind == EntryKind::File {
             "file"
         } else {
             "directory"
         };
-        let workdir = selected.session.workdir().id().as_str().to_owned();
-        let validator = self.wrap_validator(
-            &alias,
-            selected.generation,
-            &workdir,
-            &observation.validator,
-        );
+        let validator = self.wrap_validator(alias, generation, workdir, &observation.validator);
         let handler = Arc::new(FileHandler {
             provider: self.clone(),
-            alias,
-            generation: selected.generation,
-            workdir,
+            alias: alias.into(),
+            generation,
+            workdir: workdir.into(),
             target,
             observation,
         });
-        let family = format!("{kind}-g{}", selected.generation);
-        Ok(Some(self.project(
-            path,
-            descriptor(ops),
-            validator,
-            &family,
-            handler,
-        )))
+        let family = format!("{kind}-g{generation}");
+        let mut projection = self.project(path, descriptor(ops), validator, &family, handler);
+        // Keep path-qualified local names (permissions/capabilities may differ
+        // per target) while binding every content Interface to its live entrance.
+        projection.interface.scope = checkout_root(alias);
+        projection.interface.name.push_str("#target/");
+        projection.interface.name.push_str(&encode_identity(path));
+        projection.object.interfaces = vec![projection.interface.clone()];
+        Some(projection)
     }
+
     fn operations(&self, o: &CheckoutObservation) -> Vec<OperationDeclaration> {
         let mut ops = Vec::new();
         let has = |cap, name| o.capabilities.supports(cap) && self.may_allow(name);
@@ -261,6 +319,13 @@ impl Provider {
                 ));
             }
         } else {
+            if has(Cap::Read, "List") {
+                ops.push(operation(
+                    "list",
+                    "List direct contents, directories first then lexical path. Bounded live pages, not a snapshot; pass returned after to continue. Limit defaults to 100, maximum 1000. List permission and provider Read/enumeration authority apply. Includes hidden/ignored names per provider List policy, unlike Glob/Grep.",
+                    vec![parameter("limit", false, TypeExpr::Integer), parameter("after", false, cursor_type())],
+                ));
+            }
             if has(Cap::Glob, "Glob") {
                 ops.push(operation(
                     "glob",
@@ -288,6 +353,15 @@ impl Provider {
 
 #[async_trait]
 impl WipSubtreeProvider for Arc<Provider> {
+    fn interface_target(&self, reference: &wip_protocol::InterfaceReference) -> Option<String> {
+        if reference.scope == ROOT {
+            return Some(ROOT.into());
+        }
+        let (_, encoded) = reference.name.rsplit_once("#target/")?;
+        let path = decode_identity(encoded)?;
+        let (alias, _) = self.parse(&path)?;
+        (reference.scope == checkout_root(&alias)).then_some(path)
+    }
     async fn publication(&self, path: &str) -> Result<Option<WipPublication>, ProtocolError> {
         let _permit = tokio::time::timeout(self.deadline, self.permits.acquire())
             .await
@@ -298,7 +372,7 @@ impl WipSubtreeProvider for Arc<Provider> {
                 )
             })?
             .map_err(|_| error(ProtocolErrorCode::Internal, "checkout unavailable"))?;
-        tokio::time::timeout(self.deadline, self.node(path))
+        tokio::time::timeout(self.deadline, self.publication_inner(path))
             .await
             .map_err(|_| {
                 error(
@@ -306,7 +380,6 @@ impl WipSubtreeProvider for Arc<Provider> {
                     "checkout observation deadline",
                 )
             })?
-            .and_then(|projection| projection.map(WipPublication::self_scoped).transpose())
     }
     async fn children(&self, path: &str) -> Result<Vec<String>, ProtocolError> {
         let _permit = tokio::time::timeout(self.deadline, self.permits.acquire())
@@ -354,70 +427,9 @@ impl Provider {
             }
             return Ok(paths);
         }
-        let Some((alias, target)) = self.parse(path) else {
-            return Err(error(ProtocolErrorCode::NotFound, "checkout not published"));
-        };
-        let selected = self
-            .router
-            .resolve(Some(&alias))
-            .map_err(|_| error(ProtocolErrorCode::NotFound, "attachment expired"))?;
-        let o = selected
-            .session
-            .checkout_observe(target.clone())
-            .await
-            .map_err(|_| error(ProtocolErrorCode::NotFound, "checkout not published"))?;
-        if o.kind != EntryKind::Directory {
-            return Ok(Vec::new());
-        }
-        if !self.allowed(
-            "Glob",
-            &json!({"target_workdir":alias,"path":target.as_str(),"pattern":"*"}),
-        ) && !self.allowed(
-            "Grep",
-            &json!({"target_workdir":alias,"path":target.as_str(),"pattern":""}),
-        ) {
-            return Ok(Vec::new());
-        }
-        let listing = selected
-            .session
-            .checkout_search(workdir::CheckoutSearchRequest::new(
-                workdir::CheckoutSearchOperation::List(ListRequest {
-                    path: target,
-                    limit: MAX_NODES + 1,
-                }),
-            ))
-            .await
-            .map_err(|_| error(ProtocolErrorCode::Internal, "checkout listing unavailable"))?;
-        let workdir::CheckoutSearchResult::List(listing) = listing else {
-            return Err(error(
-                ProtocolErrorCode::Internal,
-                "mismatched checkout listing",
-            ));
-        };
-        if listing.truncated || listing.entries.len() > MAX_NODES {
-            return Err(error(
-                ProtocolErrorCode::ResourceLimitExceeded,
-                "directory observation exceeds node bound",
-            ));
-        }
-        let mut paths = Vec::new();
-        for entry in listing.entries {
-            let p = object_path(&alias, &entry.path);
-            if self.node(&p).await?.is_some() {
-                paths.push(p);
-            }
-        }
-        if !self
-            .router
-            .resolve(Some(&alias))
-            .is_ok_and(|current| current.generation == selected.generation)
-        {
-            return Err(error(
-                ProtocolErrorCode::ValidatorMismatch,
-                "attachment changed",
-            ));
-        }
-        Ok(paths)
+        // Contents are addressable, but never indexable. Host observe already
+        // resolves the requested Object; no provider content List is needed.
+        Ok(Vec::new())
     }
 }
 
@@ -501,6 +513,7 @@ impl WipOperationHandler for FileHandler {
         context: WipCallContext,
     ) -> Result<WipOperationOutput, WipOperationError> {
         let tool = match operation {
+            "list" => "List",
             "read" => "Read",
             "edit" => "Edit",
             "write" => "Write",
@@ -528,7 +541,34 @@ impl WipOperationHandler for FileHandler {
         }
         let mut permission_input = args.clone();
         permission_input.insert("target_workdir".into(), json!(self.alias));
-        if tool == "Glob" || tool == "Grep" {
+        if tool == "List" {
+            permission_input.insert("path".into(), json!(self.target.as_str()));
+            if let Some(after) = args.get_mut("after") {
+                let entry = after.get("entry").and_then(Json::as_str).ok_or_else(|| {
+                    protocol(
+                        ProtocolErrorCode::InvalidArguments,
+                        "after requires an entry",
+                    )
+                })?;
+                let Some((alias, path)) = self.provider.parse(entry) else {
+                    return Err(protocol(
+                        ProtocolErrorCode::InvalidArguments,
+                        "invalid continuation entry",
+                    ));
+                };
+                if alias != self.alias || object_path(&alias, &path) != entry {
+                    return Err(protocol(
+                        ProtocolErrorCode::InvalidArguments,
+                        "continuation must belong to this checkout",
+                    ));
+                }
+                let fields = after.as_object_mut().ok_or_else(|| {
+                    protocol(ProtocolErrorCode::InvalidArguments, "invalid continuation")
+                })?;
+                fields.remove("entry");
+                fields.insert("path".into(), json!(path.as_str()));
+            }
+        } else if tool == "Glob" || tool == "Grep" {
             let relative = args.get("path").and_then(Json::as_str).unwrap_or("");
             permission_input.insert("path".into(), json!(join_under(&self.target, relative)?));
         } else if tool == "Create" {
@@ -588,49 +628,49 @@ impl WipOperationHandler for FileHandler {
             }
         })?
         .map_err(map_tool_error)?;
-        let links = tokio::time::timeout_at(deadline, async {
-            let mut links = Vec::new();
-            self.check_connection()?;
-            for path in &result.paths {
-                let route = object_path(&self.alias, path);
-                // References are validated only against the captured live connection.
-                if self
-                    .provider
-                    .node(&route)
-                    .await
-                    .map_err(WipOperationError::Protocol)?
-                    .is_some()
-                {
-                    links.push(json!({"path":route}));
-                }
-                self.check_connection()?;
+        // Coordinates are not observations. Never prefetch unselected results;
+        // a vanished entry cannot fail a successful discovery Operation.
+        self.check_connection()?;
+        let mut fields = BTreeMap::from([("summary".into(), Value::String(result.output.summary))]);
+        if let Some(content) = result.output.content {
+            fields.insert("content".into(), Value::String(content));
+        }
+        let items = if let Some(listing) = result.listing {
+            fields.insert("truncated".into(), Value::Boolean(listing.truncated));
+            if let Some(after) = listing.next_after {
+                fields.insert("after".into(), cursor_value(&self.alias, &after));
             }
-            Ok::<_, WipOperationError>(links)
-        })
-        .await
-        .map_err(|_| {
-            if matches!(tool, "Write" | "Edit" | "Create") {
-                WipOperationError::OutcomeUnknown("completed operation response deadline".into())
-            } else {
-                protocol(
-                    ProtocolErrorCode::ResourceLimitExceeded,
-                    "checkout result deadline",
-                )
-            }
-        })?
-        .map_err(|e| {
-            if matches!(tool, "Write" | "Edit" | "Create") {
-                WipOperationError::OutcomeUnknown("completed operation response unavailable".into())
-            } else {
-                e
-            }
-        })?;
-        let value = json_to_wip(
-            &json!({"summary":result.output.summary,"content":result.output.content,"items":links}),
-        )
-        .map_err(|_| {
-            WipOperationError::OutcomeUnknown("invalid completed file operation result".into())
-        })?;
+            listing
+                .entries
+                .into_iter()
+                .map(|item| {
+                    Value::Record(BTreeMap::from([
+                        (
+                            "entry".into(),
+                            Value::String(object_path(&self.alias, &item.path)),
+                        ),
+                        ("kind".into(), Value::String(kind_name(item.kind).into())),
+                        (
+                            "size".into(),
+                            Value::Integer(i64::try_from(item.size).unwrap_or(i64::MAX)),
+                        ),
+                    ]))
+                })
+                .collect()
+        } else {
+            result
+                .paths
+                .iter()
+                .map(|path| {
+                    Value::Record(BTreeMap::from([(
+                        "entry".into(),
+                        Value::String(object_path(&self.alias, path)),
+                    )]))
+                })
+                .collect()
+        };
+        fields.insert("items".into(), Value::List(items));
+        let value = Value::Record(fields);
         if let Some(raw) = result.validator {
             Ok(WipOperationOutput::native_with_validator(
                 value,
@@ -693,10 +733,70 @@ fn operation(
         parameters,
         returns: ReturnDeclaration {
             documentation: None,
-            r#type: TypeExpr::Json,
+            r#type: result_type(name == "list"),
         },
     }
 }
+fn field(name: &str, required: bool, r#type: TypeExpr) -> FieldDeclaration {
+    FieldDeclaration {
+        name: name.into(),
+        required,
+        documentation: None,
+        r#type,
+    }
+}
+fn cursor_type() -> TypeExpr {
+    TypeExpr::Record {
+        fields: vec![
+            field("kind", true, TypeExpr::String),
+            field("entry", true, TypeExpr::Entry),
+        ],
+    }
+}
+fn result_type(list: bool) -> TypeExpr {
+    let mut item = vec![field("entry", true, TypeExpr::Entry)];
+    if list {
+        item.extend([
+            field("kind", true, TypeExpr::String),
+            field("size", true, TypeExpr::Integer),
+        ]);
+    }
+    let mut fields = vec![
+        field("summary", true, TypeExpr::String),
+        field("content", false, TypeExpr::String),
+        field(
+            "items",
+            true,
+            TypeExpr::List {
+                items: Box::new(TypeExpr::Record { fields: item }),
+            },
+        ),
+    ];
+    if list {
+        fields.extend([
+            field("truncated", true, TypeExpr::Boolean),
+            field("after", false, cursor_type()),
+        ]);
+    }
+    TypeExpr::Record { fields }
+}
+fn kind_name(kind: EntryKind) -> &'static str {
+    match kind {
+        EntryKind::Directory => "directory",
+        EntryKind::File => "file",
+        EntryKind::Symlink => "symlink",
+        EntryKind::Other => "other",
+    }
+}
+fn cursor_value(alias: &str, cursor: &workdir::ListCursor) -> Value {
+    Value::Record(BTreeMap::from([
+        ("kind".into(), Value::String(kind_name(cursor.kind).into())),
+        (
+            "entry".into(),
+            Value::String(object_path(alias, &cursor.path)),
+        ),
+    ]))
+}
 fn descriptor(operations: Vec<OperationDeclaration>) -> InterfaceDescriptor {
     InterfaceDescriptor{format:INTERFACE_FORMAT_V1.into(),documentation:Some(Documentation{summary:"AI-oriented attached Workdir operations".into(),details:Some("The route binds attachment and target. Line offsets are zero-based; text output numbers lines from one. Read-before-write and provider validators are both required. Paths are Worldspace references, never host paths.".into())}),types:Vec::new(),operations}
 }

crates/worker/src/checkout_http_tests.rs

@@ -379,7 +379,7 @@ fn first_link(value: &Value) -> &str {
         panic!("expected typed link List")
     };
     assert_eq!(items.len(), 1);
-    let Value::String(path) = field(&items[0], "path") else {
+    let Value::String(path) = field(&items[0], "entry") else {
         panic!("expected typed link path")
     };
     path
@@ -669,3 +669,89 @@ async fn checkout_http_malformed_response_after_write_is_unknown_without_retry()
 async fn checkout_http_lost_response_after_write_is_unknown_without_retry() {
     watchdog(committed_write_unknown(WriteResponseFault::LostBody)).await;
 }
+
+#[tokio::test]
+async fn remote_checkout_list_pages_and_direct_inspect_read_use_provider_without_tree_enumeration()
+{
+    use super::checkout_wip_tests::{json_content, model_inspect, model_invoke, model_operation};
+    let fixture = HttpFixture::new(WorkdirSessionCapabilities::READ_ONLY).await;
+    std::fs::create_dir(fixture.dir.path().join("empty")).unwrap();
+    let remote = fixture.open(TOKEN).await.unwrap();
+    let (runtime, _) = native_runtime(Arc::new(remote));
+    let runtime = Arc::new(runtime);
+    let tree = json_content(
+        model_operation(
+            runtime.clone(),
+            "Tree",
+            json!({"path":"/checkouts", "depth":8}),
+        )
+        .await
+        .unwrap(),
+    );
+    assert_eq!(tree["tree"]["children"][0]["children"], json!([]));
+    assert!(
+        !fixture
+            .state
+            .operations
+            .lock()
+            .unwrap()
+            .iter()
+            .any(|op| matches!(
+                op,
+                WorkdirSessionOperation::CheckoutSearch(_) | WorkdirSessionOperation::List(_)
+            ))
+    );
+    let entrance = model_inspect(runtime.clone(), "/checkouts/main").await;
+    let first = json_content(
+        model_invoke(runtime.clone(), &entrance, "list", json!({"limit":1}))
+            .await
+            .unwrap(),
+    );
+    assert_eq!(first["items"][0]["entry"], "/checkouts/main/empty");
+    assert_eq!(first["items"][0]["kind"], "directory");
+    assert_eq!(first["truncated"], true);
+    let next = json_content(
+        model_invoke(
+            runtime.clone(),
+            &entrance,
+            "list",
+            json!({"limit":1,"after":first["after"]}),
+        )
+        .await
+        .unwrap(),
+    );
+    assert_eq!(next["items"][0]["entry"], "/checkouts/main/src");
+    assert_eq!(next["truncated"], false);
+    let deep = model_inspect(runtime.clone(), "/checkouts/main/src/deep").await;
+    for (operation, args) in [
+        ("list", json!({})),
+        ("glob", json!({"pattern":"*.txt"})),
+        (
+            "grep",
+            json!({"pattern":"needle","output_mode":"files_with_matches"}),
+        ),
+    ] {
+        let found = json_content(
+            model_invoke(runtime.clone(), &deep, operation, args)
+                .await
+                .unwrap(),
+        );
+        let file = found["items"][0]["entry"].as_str().unwrap();
+        assert_eq!(file, "/checkouts/main/src/deep/a.txt");
+        let inspected = model_inspect(runtime.clone(), file).await;
+        let read = json_content(
+            model_invoke(runtime.clone(), &inspected, "read", json!({}))
+                .await
+                .unwrap(),
+        );
+        assert!(read["content"].as_str().unwrap().contains("needle"));
+        fixture.public_output(&found.to_string());
+    }
+    let operations = fixture.state.operations.lock().unwrap();
+    assert!(operations.iter().any(|op| matches!(op, WorkdirSessionOperation::CheckoutSearch(r) if matches!(&r.operation, CheckoutSearchOperation::List(l) if l.after.is_some()))));
+    assert!(
+        !operations
+            .iter()
+            .any(|op| matches!(op, WorkdirSessionOperation::List(_)))
+    );
+}

crates/worker/src/checkout_race_tests.rs

@@ -13,9 +13,9 @@ use tempfile::TempDir;
 use workdir::{
     CheckoutRequest, CheckoutResult, CommandHandle, CommandOutput, CommandOutputRequest,
     CommandRequest, CommandStatus, EditRequest, EditResult, GlobRequest, GlobResult, GrepRequest,
-    GrepResult, ListResult, LocalWorkdirSession, ReadBytesRequest, ReadBytesResult, ReadRequest,
-    ReadResult, StatRequest, StatResult, Workdir, WorkdirAttachmentAlias, WorkdirError,
-    WorkdirScopeAuthorizationRequest, WorkdirScopeOverlapRequest, WorkdirSession,
+    GrepResult, ListRequest, ListResult, LocalWorkdirSession, ReadBytesRequest, ReadBytesResult,
+    ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirAttachmentAlias,
+    WorkdirError, WorkdirScopeAuthorizationRequest, WorkdirScopeOverlapRequest, WorkdirSession,
     WorkdirSessionCapabilities, WorkdirSessionHandle, WriteRequest, WriteResult,
 };
 
@@ -28,6 +28,7 @@ struct HookedSession {
     observe_hook: Mutex<Option<ObserveHook>>,
     stall_path: Mutex<Option<WorkdirPath>>,
     stalled_observations: AtomicUsize,
+    searches: AtomicUsize,
 }
 
 impl std::fmt::Debug for HookedSession {
@@ -52,6 +53,7 @@ impl HookedSession {
             observe_hook: Mutex::new(None),
             stall_path: Mutex::new(None),
             stalled_observations: AtomicUsize::new(0),
+            searches: AtomicUsize::new(0),
         })
     }
 
@@ -123,6 +125,7 @@ impl WorkdirSession for HookedSession {
         &self,
         request: workdir::CheckoutSearchRequest,
     ) -> Result<workdir::CheckoutSearchResult, WorkdirError> {
+        self.searches.fetch_add(1, Ordering::SeqCst);
         self.local.checkout_search(request).await
     }
 
@@ -484,7 +487,7 @@ async fn collection_list_and_root_enumeration_bound_stalled_observation_and_rele
 }
 
 #[tokio::test]
-async fn grep_post_search_link_observation_is_deadline_bounded_and_releases_permit() {
+async fn grep_returns_typed_entry_without_observing_stalled_result() {
     let dir = TempDir::new().unwrap();
     std::fs::write(dir.path().join("a.txt"), "needle\n").unwrap();
     let session = HookedSession::new(&dir);
@@ -499,8 +502,8 @@ async fn grep_post_search_link_observation_is_deadline_bounded_and_releases_perm
     let root = checkout_root("main");
     let projection = p.node(&root).await.unwrap().unwrap();
 
-    // Directory observations and the normal local grep/checkout execution can
-    // complete. Only publishing the typed search result path gets stuck.
+    // Directory/search can complete. The result path must never be observed
+    // during result publication, even when it would block indefinitely.
     session.stall_observation(Some(WorkdirPath::new("a.txt").unwrap()));
     let arguments = BTreeMap::from([
         ("pattern".into(), Value::String("needle".into())),
@@ -514,13 +517,17 @@ async fn grep_post_search_link_observation_is_deadline_bounded_and_releases_perm
         projection.handler.call("grep", &arguments, context()),
     )
     .await
-    .expect("post-search link observations must share the operation deadline");
-    assert_deadline(result, "checkout result deadline");
-    assert_eq!(session.stalled_observations.load(Ordering::SeqCst), 1);
+    .expect("completed search must return without acquiring its entries");
+    let output = result.unwrap_or_else(|_| panic!("successful search must not observe each entry"));
+    assert_eq!(
+        wip_to_json(&output.value).unwrap()["items"],
+        json!([{"entry": format!("{root}/a.txt")}])
+    );
+    assert_eq!(session.stalled_observations.load(Ordering::SeqCst), 0);
     assert_permit_released(&p);
 
-    // A successful retry both checks cancellation cleanup and proves that the
-    // delegated search produces the typed path consumed by post-processing.
+    // A second explicit search produces the same typed coordinate, without
+    // acquiring it even when it could resolve successfully.
     session.stall_observation(None);
     let result = tokio::time::timeout(
         Duration::from_secs(2),
@@ -533,6 +540,33 @@ async fn grep_post_search_link_observation_is_deadline_bounded_and_releases_perm
         Err(_) => panic!("grep must succeed once the link observation is unstalled"),
     };
     let value = wip_to_json(&output.value).unwrap();
-    assert_eq!(value["items"], json!([{"path": format!("{root}/a.txt")}]));
+    assert_eq!(value["items"], json!([{"entry": format!("{root}/a.txt")}]));
     assert_permit_released(&p);
 }
+
+#[tokio::test]
+async fn checkout_content_children_never_call_provider_list_even_for_deep_paths() {
+    let dir = TempDir::new().unwrap();
+    std::fs::create_dir_all(dir.path().join("nested/deep")).unwrap();
+    for n in 0..1100 {
+        std::fs::write(dir.path().join(format!("file-{n}")), "fixture").unwrap();
+    }
+    let session = HookedSession::new(&dir);
+    let router = Arc::new(WorkdirSessionRouter::new());
+    router
+        .attach(
+            WorkdirAttachmentAlias::new("main").unwrap(),
+            session.clone(),
+        )
+        .unwrap();
+    let subtree = Arc::new(provider(router));
+    assert_eq!(
+        subtree.children(ROOT).await.unwrap(),
+        vec![checkout_root("main")]
+    );
+    for path in ["/checkouts/main", "/checkouts/main/nested/deep"] {
+        assert!(subtree.publication(path).await.unwrap().is_some());
+        assert!(subtree.children(path).await.unwrap().is_empty());
+    }
+    assert_eq!(session.searches.load(Ordering::SeqCst), 0);
+}

crates/worker/src/checkout_wip_tests.rs

@@ -113,7 +113,7 @@ async fn checkout_wip_search_preserves_root_and_nested_ignore_rules_and_links()
                 .as_array()
                 .unwrap()
                 .iter()
-                .map(|item| item["path"].as_str().unwrap().to_string())
+                .map(|item| item["entry"].as_str().unwrap().to_string())
                 .collect::<Vec<_>>();
             // Grep retains traversal order; Glob sorts its result paths.
             paths.sort();
@@ -159,8 +159,8 @@ async fn checkout_wip_native_roundtrip_search_read_edit_write_create() {
     .await
     .unwrap();
     let grep_result: Json = serde_json::from_str(grep.content.as_deref().unwrap()).unwrap();
-    assert_eq!(grep_result["items"][0]["path"], file);
-    let followed = grep_result["items"][0]["path"].as_str().unwrap();
+    assert_eq!(grep_result["items"][0]["entry"], file);
+    let followed = grep_result["items"][0]["entry"].as_str().unwrap();
     assert_eq!(followed, file);
     assert!(
         call(
@@ -588,8 +588,9 @@ async fn checkout_wip_scoped_child_search_and_revoke_cannot_expose_siblings() {
         .observe_live("/checkouts/child", 2)
         .await
         .unwrap();
-    assert_eq!(observed.children.unwrap().len(), 1);
+    assert!(observed.children.unwrap().is_empty());
     for (op, args) in [
+        ("list", json!({"limit":1})),
         ("glob", json!({"pattern":"**/*"})),
         (
             "grep",
@@ -654,3 +655,339 @@ async fn checkout_wip_scoped_child_search_and_revoke_cannot_expose_siblings() {
         "changed"
     );
 }
+
+pub(super) fn json_content(output: ToolOutput) -> Json {
+    serde_json::from_str(output.content.as_deref().unwrap()).unwrap()
+}
+pub(super) async fn model_operation(
+    r: Arc<WipRuntime>,
+    name: &str,
+    args: Json,
+) -> Result<ToolOutput, ToolError> {
+    let (_, tool) = wip_tool_definitions(r)
+        .into_iter()
+        .map(|definition| definition())
+        .find(|(meta, _)| meta.name == name)
+        .unwrap();
+    tool.execute(&args.to_string(), Default::default()).await
+}
+pub(super) async fn model_inspect(r: Arc<WipRuntime>, path: &str) -> Json {
+    json_content(
+        model_operation(r, "Inspect", json!({"path":path}))
+            .await
+            .unwrap(),
+    )
+}
+pub(super) async fn model_invoke(
+    r: Arc<WipRuntime>,
+    inspected: &Json,
+    operation: &str,
+    arguments: Json,
+) -> Result<ToolOutput, ToolError> {
+    model_operation(r, "Invoke", json!({"path":inspected["path"], "interface":inspected["interfaces"][0]["reference"], "operation":operation, "arguments":arguments})).await
+}
+
+#[tokio::test]
+async fn checkout_depth_boundaries_are_independent_of_empty_small_or_large_contents() {
+    for count in [0, 2, 1100] {
+        let dir = TempDir::new().unwrap();
+        std::fs::create_dir_all(dir.path().join("nested/deep")).unwrap();
+        for n in 0..count {
+            std::fs::write(dir.path().join(format!("file-{n}")), "fixture").unwrap();
+        }
+        let router = Arc::new(WorkdirSessionRouter::new());
+        router
+            .attach(
+                WorkdirAttachmentAlias::new("main").unwrap(),
+                session(&dir, true),
+            )
+            .unwrap();
+        let r = runtime(router, None);
+        let one = r.host.observe_live("/checkouts", 1).await.unwrap();
+        assert!(one.children.unwrap()[0].children.is_none());
+        for depth in [2, 32] {
+            let observation = r.host.observe_live("/checkouts", depth).await.unwrap();
+            let entrances = observation.children.unwrap();
+            assert_eq!(entrances.len(), 1);
+            assert_eq!(entrances[0].children, Some(vec![]));
+        }
+        let world = r.host.observe_live("/", 32).await.unwrap();
+        assert_eq!(
+            world.children.unwrap()[0].children.as_ref().unwrap()[0].children,
+            Some(vec![])
+        );
+        for path in ["/checkouts/main", "/checkouts/main/nested/deep"] {
+            assert!(
+                r.host
+                    .observe_live(path, 0)
+                    .await
+                    .unwrap()
+                    .children
+                    .is_none()
+            );
+            assert_eq!(
+                r.host.observe_live(path, 1).await.unwrap().children,
+                Some(vec![])
+            );
+        }
+    }
+}
+
+#[tokio::test]
+async fn checkout_model_list_glob_grep_entries_are_lazy_and_never_become_tree_edges() {
+    let dir = TempDir::new().unwrap();
+    std::fs::create_dir_all(dir.path().join("empty")).unwrap();
+    std::fs::create_dir_all(dir.path().join("nested/deep")).unwrap();
+    std::fs::write(dir.path().join("nested/deep/a.txt"), "needle\n").unwrap();
+    std::fs::write(dir.path().join("z.txt"), "needle\n").unwrap();
+    let router = Arc::new(WorkdirSessionRouter::new());
+    router
+        .attach(
+            WorkdirAttachmentAlias::new("main").unwrap(),
+            session(&dir, true),
+        )
+        .unwrap();
+    let r = Arc::new(runtime(router, None));
+    let entrance = model_inspect(r.clone(), "/checkouts/main").await;
+    assert_eq!(entrance["path"], "/checkouts/main");
+    assert_eq!(
+        entrance["interfaces"][0]["reference"]["scope"],
+        "/checkouts/main"
+    );
+    let projection = r
+        .host
+        .projection_live("/checkouts/main")
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        entrance["object_signature"],
+        wip_text_view::render_object(&projection.object).unwrap()
+    );
+    assert_eq!(
+        entrance["interfaces"][0]["signature"],
+        wip_text_view::render_interface(&wip_text_view::Interface {
+            reference: &projection.interface,
+            descriptor: &projection.descriptor
+        })
+        .unwrap()
+    );
+    assert!(
+        entrance["interfaces"][0]["signature"]
+            .as_str()
+            .unwrap()
+            .contains("entry")
+    );
+    let mut after = None;
+    let mut items = vec![];
+    loop {
+        let mut args = json!({"limit":1});
+        if let Some(cursor) = after {
+            args["after"] = cursor;
+        }
+        let page = json_content(
+            model_invoke(r.clone(), &entrance, "list", args)
+                .await
+                .unwrap(),
+        );
+        assert_eq!(page["items"].as_array().unwrap().len(), 1);
+        items.push(page["items"][0].clone());
+        if page["truncated"] == false {
+            assert!(page.get("after").is_none());
+            break;
+        }
+        after = Some(page["after"].clone());
+    }
+    assert_eq!(
+        items
+            .iter()
+            .map(|item| item["entry"].clone())
+            .collect::<Vec<_>>(),
+        vec![
+            json!("/checkouts/main/empty"),
+            json!("/checkouts/main/nested"),
+            json!("/checkouts/main/z.txt")
+        ]
+    );
+    assert_eq!(items[0]["kind"], "directory");
+    assert_eq!(items[2]["kind"], "file");
+    let empty = model_inspect(r.clone(), items[0]["entry"].as_str().unwrap()).await;
+    assert_eq!(
+        json_content(
+            model_invoke(r.clone(), &empty, "list", json!({}))
+                .await
+                .unwrap()
+        )["items"],
+        json!([])
+    );
+    let deep = model_inspect(r.clone(), "/checkouts/main/nested/deep").await;
+    assert_eq!(
+        deep["interfaces"][0]["reference"]["scope"],
+        "/checkouts/main"
+    );
+    for (operation, args) in [
+        ("list", json!({})),
+        ("glob", json!({"pattern":"*.txt"})),
+        (
+            "grep",
+            json!({"pattern":"needle", "output_mode":"files_with_matches"}),
+        ),
+    ] {
+        let found = json_content(
+            model_invoke(r.clone(), &deep, operation, args)
+                .await
+                .unwrap(),
+        );
+        let path = found["items"][0]["entry"].as_str().unwrap();
+        assert_eq!(path, "/checkouts/main/nested/deep/a.txt");
+        // No result-wide eager acquisition by Host or Client.
+        if operation == "list" {
+            let state = r.state.lock().unwrap();
+            assert!(state.client.object(&state.session, path).is_none());
+        }
+        let inspected = model_inspect(r.clone(), path).await;
+        let read = json_content(
+            model_invoke(r.clone(), &inspected, "read", json!({}))
+                .await
+                .unwrap(),
+        );
+        assert!(read["content"].as_str().unwrap().contains("needle"));
+    }
+    let tree = json_content(
+        model_operation(r.clone(), "Tree", json!({"path":"/checkouts", "depth":8}))
+            .await
+            .unwrap(),
+    );
+    assert_eq!(tree["tree"]["children"][0]["children"], json!([]));
+    std::fs::remove_file(dir.path().join("nested/deep/a.txt")).unwrap();
+    assert!(
+        model_operation(
+            r,
+            "Inspect",
+            json!({"path":"/checkouts/main/nested/deep/a.txt", "refresh":true})
+        )
+        .await
+        .is_err()
+    );
+}
+
+#[tokio::test]
+async fn checkout_list_permission_cannot_be_bypassed_with_glob_or_command_identity() {
+    let dir = TempDir::new().unwrap();
+    std::fs::write(dir.path().join("a.txt"), "fixture").unwrap();
+    let router = Arc::new(WorkdirSessionRouter::new());
+    router
+        .attach(
+            WorkdirAttachmentAlias::new("main").unwrap(),
+            session(&dir, true),
+        )
+        .unwrap();
+    let r = runtime(
+        router,
+        Some(ToolPermissionConfig {
+            default_action: ToolPermissionAction::Allow,
+            rules: vec![manifest::ToolPermissionRule {
+                tool: "List".into(),
+                pattern: "*".into(),
+                action: ToolPermissionAction::Deny,
+            }],
+        }),
+    );
+    let p = r
+        .host
+        .projection_live("/checkouts/main")
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(!p.descriptor.operations.iter().any(|op| op.name == "list"));
+    assert!(p.descriptor.operations.iter().any(|op| op.name == "glob"));
+    let denied = p
+        .handler
+        .call(
+            "list",
+            &BTreeMap::new(),
+            WipCallContext {
+                execution: Default::default(),
+                security_context: "fixture".into(),
+            },
+        )
+        .await;
+    assert!(
+        matches!(denied, Err(WipOperationError::Protocol(e)) if e.code == ProtocolErrorCode::PermissionDenied)
+    );
+}
+
+#[tokio::test]
+async fn checkout_ancestor_scope_rejects_siblings_prefixes_refs_and_old_connection_before_read() {
+    let dir = TempDir::new().unwrap();
+    std::fs::write(dir.path().join("a.txt"), "same\n").unwrap();
+    let router = Arc::new(WorkdirSessionRouter::new());
+    let main = WorkdirAttachmentAlias::new("main").unwrap();
+    router.attach(main.clone(), session(&dir, true)).unwrap();
+    router
+        .attach(
+            WorkdirAttachmentAlias::new("main-extra").unwrap(),
+            session(&dir, true),
+        )
+        .unwrap();
+    let r = runtime(router.clone(), None);
+    let path = "/checkouts/main/a.txt";
+    let p = r.host.projection_live(path).await.unwrap().unwrap();
+    let fetched = r.host.fetch_interface_live(&p.interface).await.unwrap();
+    assert!(
+        fetched.scope_ref.is_none(),
+        "ref-less publication is legitimate"
+    );
+    // Ordinary content changes do not terminate the entrance's Interface lifetime.
+    std::fs::write(dir.path().join("new.txt"), "fixture").unwrap();
+    assert_eq!(
+        r.host
+            .fetch_interface_live(&p.interface)
+            .await
+            .unwrap()
+            .interface,
+        p.interface
+    );
+    let request = |reference, scope_ref| CallOperationRequest {
+        target: wip_protocol::Target {
+            path: path.into(),
+            validator: None,
+        },
+        interface: wip_protocol::InterfaceTarget {
+            reference,
+            scope_ref,
+            validator: None,
+        },
+        operation: "read".into(),
+        arguments: BTreeMap::new(),
+    };
+    for scope in ["/checkouts/main-extra", "/checkouts/main/a"] {
+        let mut wrong = p.interface.clone();
+        wrong.scope = scope.into();
+        assert!(r.host.fetch_interface_live(&wrong).await.is_err());
+        let forged = request(wrong, None);
+        let result = r
+            .host
+            .call(
+                forged,
+                WipCallContext {
+                    execution: Default::default(),
+                    security_context: "fixture".into(),
+                },
+            )
+            .await;
+        assert!(
+            matches!(result,Err(WipOperationError::Protocol(e)) if e.code == ProtocolErrorCode::InterfaceMismatch)
+        );
+    }
+    let forged = request(p.interface.clone(), Some("old".into()));
+    assert!(
+        matches!(r.host.call(forged, WipCallContext { execution:Default::default(),security_context:"fixture".into() }).await, Err(WipOperationError::Protocol(e)) if e.code == ProtocolErrorCode::InterfaceMismatch)
+    );
+    router.detach(&main).await.unwrap();
+    router.attach(main, session(&dir, true)).unwrap();
+    let forged = request(p.interface.clone(), None);
+    assert!(
+        matches!(r.host.call(forged, WipCallContext { execution:Default::default(),security_context:"fixture".into() }).await, Err(WipOperationError::Protocol(e)) if e.code == ProtocolErrorCode::InterfaceMismatch)
+    );
+}

crates/worker/src/fs_view.rs

@@ -143,6 +143,7 @@ impl WorkerFsView {
                 .list(ListRequest {
                     path: logical.clone(),
                     limit: DIR_FILE_REF_ENTRY_LIMIT,
+                    after: None,
                 })
                 .await
                 .map_err(ToolsError::from)
@@ -222,6 +223,7 @@ impl WorkerFsView {
             .list(ListRequest {
                 path: parent,
                 limit: COMPLETION_LIMIT,
+                after: None,
             })
             .await
         else {

crates/workspace-server/src/server.rs

@@ -54114,6 +54114,7 @@ mod tests {
             WorkdirSessionOperation::List(workdir::ListRequest {
                 path: workdir::WorkdirPath::root(),
                 limit: 10,
+                after: None,
             }),
         )
         .await