feat(lsp): D3 --- the file watcher stops sleeping and walks once per scan (#233)

Implements docs/lsp-file-watch-d3-framing.md revision 4, approved
2026-08-11 with the four rulings adopted as proposed: the honest bar
(absent at idle, one attributable job per concurrently due group), no
exclusions by default, server root_uri -> cwd -> attachment fallback,
and constants rather than config keys.

pmacs.fs.walk_tree: the whole recursive tree as ONE cancellable job
(JobKind::FsWalkTree, reply reuses ReplyKind::ReadDir --- identical
payload shape, so the Lua boundary needs no second conversion). Names
are base-relative; symlinks recorded, never traversed; an unreadable
subdirectory skips its subtree (scan_tree's pcall behaviour); only the
root failing to open fails the walk; the cancel token is polled once
per directory. Eight Rust unit tests, including flat-directory entry
parity with read_dir_blocking and the two review-round cancellation
cases (empty-tree pre-cancel; mid-walk via the cfg(test) entry hook).

The watcher itself is rewritten as the framed group scheduler. No
sleeps anywhere: one process.after-tick subscription (installed once
and guarded --- pmacs.hook.remove does not exist) drives every
(server, base) group's deadline off monotonic_ms, autosave's Q#AS2
idiom. The old design held one pool thread per sleeping watcher and
allocated 1 sleep + D read_dir jobs per watcher per tick --- 1,326
per tick for rust-analyzer's six watchers on this 220-directory
checkout. At idle there is now NO running job, which is also the
strongest witness in the suite: activity_summary settles to None, and
that assertion is unwritable under the old design.

The scheduler is the framing's state machine, all three review rounds
included: single-flight per group with generation-checked completions;
deadlines advanced from completion; the round-3 three-arm completion
partition (success / stale-or-retired / live non-success, with the
failure latch and quiet cancellation); joins wake the group, queue
exactly one follow-up mid-walk, and never reset the backoff curve;
per-watcher baselines --- the first snapshot whose WALK STARTED after
the join; membership captured at scan start; per-member cancellation
recheck at emit through the preserved _after_scan_for_tests seam;
backoff 250ms x2 to a 4s cap, reset by any emitted change; retirement
cancels the in-flight walk cooperatively.

Verification: eighteen acceptance tests. The six #234 tests are
byte-unchanged and green. Ten witnesses cover the framing's plan (the
review rounds added the fallback-determinism and root-boundary pair,
making twelve):
idle absence (and never a sleep purpose), one walk job per scan on a
twelve-directory fixture, join-wakes plus the registration epoch,
queued baseline for a mid-walk join (driven by saturating the worker
pool so the walk genuinely queues), single-flight under a withheld
completion pump, retirement and rebaseline through the fake's
unregister/re-register triggers, live cancel via pmacs._async._cancel
on the queued job, live failure with the once-per-error latch and the
preserved-snapshot recovery (DELETED for the pre-failure file is only
derivable from the retained snapshot), backoff shape from seam
timestamps, and the configured-root base.

Every witness was mutation-tested. Two findings from the bites:

- Retirement is DOUBLE-ENFORCED (unregister path and post-scan sweep)
  and biting either copy alone is masked by the other; only biting
  both goes red. Kept deliberately: the sweep covers seam-cancelled
  members, the unregister path covers idle groups whose next deadline
  is seconds away.
- The first idle probe was VACUOUS: it read pmacs.async instead of
  pmacs._async, errored, and the unwrap_or_default made every sample
  read as "absent". The probe now expects rather than defaults, so a
  broken probe is a red test, not a green lie.

One environmental fact, recorded in the lane: an empty stray /tmp/.git
(since removed) made project detection root every markerless tempdir
fixture at /tmp, which under Q#D3-3 the watcher then faithfully
watched. A markerless-fixture red that looks like a watcher bug may be
an ancestor marker.

A pre-commit review round found four blockers, all fixed here:

- walk_tree checked cancellation only inside its entry loops, which an
  EMPTY tree never enters --- a pre-cancelled queued walk returned an
  empty SUCCESS, which the success arm would commit and diff into a
  deletion storm. Cancellation is now checked before opening and
  before returning, cancellation outranks a missing-root error, and a
  unit test pins both.
- The neither-root-nor-cwd attachment fallback was still pairs-order
  nondeterministic --- the exact accident D3 set out to remove, behind
  a comment claiming otherwise. It now takes the lexicographically
  smallest attachment directory. Verified at the spawn sites: every
  server spawned with an attached file gets cwd = root, so the arm is
  defensive and unreachable through production spawning --- which is
  also why it carries no through-the-server witness.
- A base at the filesystem root joined as //path (and file:////path in
  URIs). Both join sites now go through join_under, the root-aware
  idiom dired's handler already uses, and the dir-of capture for a
  root-level file ("" from the match) normalizes to "/".
- The walk-count and scan-times probes defaulted on error, so two
  broken probes could compare equal and pass the retirement witness.
  Every probe now expects --- a broken probe is a red test, the same
  correction the vacuous idle probe forced.

A second pre-commit round found three more, all fixed here:

- Mid-walk cancellation was UNWITNESSED: both Rust cancel tests
  pre-cancelled and the acceptance test cancelled a queued walk, so
  deleting the internal polls left every test green. A cfg(test)
  entry hook now flips the token at an exact entry boundary and the
  witness asserts the walk stopped NEAR it (bound on entries
  processed), which is what discriminates the polls from the
  entry/exit checks. The retirement witness now holds a walk in
  flight across the unregister and asserts the job settles cancelled.
- The "unreachable fallback" claim was WRONG: pmacs.lsp.spawn may
  omit both cwd and root_uri, and ensure_server adopts such a live
  server for markerless files (root_uri and key_uri both nil). The
  lexicographic-minimum fallback now has a through-the-server
  witness: five sibling directories, the minimum opened last ---
  five, because with two the build's hash order coincided with the
  lexicographic answer and the first-pairs bite survived.
- The root-boundary joins gained a witness through exported
  production functions (the _deliver_status pattern): the matcher and
  URI builder driven at base "/", where reverting either join_under
  call makes the anchored glob refuse //hit and the URI grow a fourth
  slash. No fixture can walk / for real.

Verification totals after both rounds: eight walk_tree unit tests,
eighteen acceptance tests (six byte-unchanged, twelve witnesses), all
mutation-verified.

No wire change, no PROTOCOL_VERSION bump; walk_tree is an fs binding.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

builtin/runtime/fs.lua

@@ -106,6 +106,7 @@ end
 
 local READ_DIR_OPTS = { supersede = true, tolerant = true }
 local STAT_OPTS = { supersede = true }
+local WALK_TREE_OPTS = { supersede = true }
 
 -- Two result shapes, chosen by `opts.tolerant` (dired Q#DR6):
 --
@@ -130,6 +131,22 @@ function fs.read_dir(path, opts)
   return build_handle(id)
 end
 
+-- walk_tree(base [, opts]) -> handle; await -> { <entry>, ... } where
+-- each entry is the read_dir shape but `name` is a BASE-RELATIVE path
+-- ("sub/dir/file.txt") and the listing covers the whole tree as ONE
+-- job (issue #233 D3). Symlinks are recorded, never traversed; an
+-- unreadable subdirectory is skipped with its subtree; only the root
+-- failing to open fails the walk. Directory entries are included
+-- (kind "dir") --- consumers that only want files filter on kind.
+function fs.walk_tree(base, opts)
+  if type(base) ~= "string" then
+    error("pmacs.fs.walk_tree: base must be a string, got " .. type(base))
+  end
+  local key = read_opts(opts, "pmacs.fs.walk_tree", WALK_TREE_OPTS)
+  local id = async_mod._dispatch_fs_walk_tree(base, key)
+  return build_handle(id)
+end
+
 function fs.stat(path, opts)
   if type(path) ~= "string" then
     error("pmacs.fs.stat: path must be a string, got " .. type(path))

builtin/runtime/lsp.lua

@@ -1910,22 +1910,33 @@ local function repull_for_attachments(sid, request_fn)
   end
 end
 
--- T M4.5 — workspace file watching (workspace/didChangeWatchedFiles).
+-- T M4.5 / issue #233 D3 — workspace file watching
+-- (workspace/didChangeWatchedFiles).
 --
 -- Servers register watchers dynamically via client/registerCapability.
--- pmacs has no kernel file-watch, so each registration runs a polling
+-- pmacs has no kernel file-watch, so watching is a polling
--- snapshot-diff coroutine: walk the base dir into a { relpath = sig }
+-- snapshot-diff — but scheduled, not slept (D3 framing, approved
--- map and, every tick, diff against the previous map to emit per-file
+-- 2026-08-11): one SCAN GROUP per (server, base) owns a retained
--- created/changed/deleted FileEvents (filtered by the glob and the
+-- { relpath = sig } snapshot, and a single `process.after-tick`
--- WatchKind bitmask), batched into one notification. Coarser than an
+-- subscription drives every group's cadence off
--- inotify bridge but accurate; a watcher self-cancels when the server
+-- `pmacs.editor.monotonic_ms` (autosave's Q#AS2 idiom). Waiting
--- dies or the capability is unregistered.
+-- allocates no job and holds no pool thread; a due group runs ONE
+-- `pmacs.fs.walk_tree` job for the whole tree, diffs in Lua, and
+-- routes per-file created/changed/deleted FileEvents through each
+-- member watcher's glob and WatchKind mask, deduped into one
+-- notification. Quiet scans back the interval off to a cap; any
+-- change resets it. Groups retire when their last member leaves or
+-- their server dies, cancelling an in-flight walk cooperatively.
 
 local FILE_WATCH_INTERVAL_MS = 250
+local FILE_WATCH_BACKOFF_CAP_MS = 4000
 
 -- file_watchers[tostring(sid)][registrationId] = list of watch records
--- ({ cancelled = bool, form = "relative"|"absolute", _sleep = handle? }),
+-- ({ cancelled, form = "relative"|"absolute", kind_mask, match_subject,
--- one per glob watcher.
+--    baseline_epoch, group }), one per glob watcher. `baseline_epoch`
+-- is nil until the first snapshot whose WALK STARTED after the record
+-- joined its group — the registration-epoch rule that keeps a joiner
+-- from receiving events for files that predate it.
 local file_watchers = {}
 
 -- WatchKind is a bitmask (Create=1, Change=2, Delete=4); test it
@@ -2029,108 +2040,328 @@ local function glob_matcher(glob)
   end
 end
 
--- Recursively list files under `base` → { relpath = sig }. `sig`
--- folds size+mtime+kind so a content/metadata change flips it.
--- Symlinks are recorded, not traversed (loop-safe). Awaits fs
--- primitives, so call from inside an async coroutine.
-local function scan_tree(base, matches)
-  local out = {}
-  local function walk(dir, rel_prefix)
-    local ok, entries = pcall(function()
-      return pmacs.fs.read_dir(dir):await()
-    end)
-    if not ok or not entries then return end
-    for _, e in ipairs(entries) do
-      local rel = (rel_prefix == "") and e.name or (rel_prefix .. "/" .. e.name)
-      if e.kind == "dir" then
-        walk(dir .. "/" .. e.name, rel)
-      elseif matches(rel) then
-        out[rel] = table.concat({
-          tostring(e.size), tostring(e.mtime),
-          tostring(e.mtime_nsec), tostring(e.kind),
-        }, "|")
-      end
-    end
-  end
-  walk(base, "")
-  return out
-end
-
 local FC_CREATED, FC_CHANGED, FC_DELETED = 1, 2, 3
 
-local function start_file_watcher(sid, base, glob, kind_mask, record)
+-- Join a base-relative path under its base. The filesystem root is
-  -- Per LSP, a plain-string glob matches the file's ABSOLUTE path,
+-- special-cased the way `dired`'s handler already spells it (the
-  -- while a RelativePattern's pattern is relative to its base — the
+-- `(dir == "/") and "" or dir` idiom): a naive `base .. "/" .. rel`
-  -- record's `form` (from resolve_watcher) picks the match subject.
+-- at `/` yields `//path` — the implementation-defined POSIX spelling
-  -- scan_tree always walks in relative terms; only the string handed
+-- — and `file:////path` once a URI wraps it.
-  -- to the matcher changes.
+local function join_under(base, rel)
-  local match_glob = glob_matcher(glob)
+  if base == "/" then return "/" .. rel end
-  local matches = match_glob
+  return base .. "/" .. rel
-  if record.form == "absolute" then
+end
-    matches = function(rel)
+
-      return match_glob(base .. "/" .. rel)
+-- Build a record's match predicate over a base-RELATIVE path. Per LSP
+-- (and #234's P1 review), a plain-string glob matches the file's
+-- ABSOLUTE path while a RelativePattern's pattern is relative to its
+-- base — the record's `form` picks the subject; the walk itself is
+-- always relative.
+local function make_matcher(base, pat, form)
+  local match_glob = glob_matcher(pat)
+  if form == "absolute" then
+    return function(rel)
+      return match_glob(join_under(base, rel))
     end
   end
-  pmacs.async(function()
+  return match_glob
-    local prev = scan_tree(base, matches)
+end
-    while not record.cancelled and server_is_live(sid) do
-      local sh = pmacs.workers.sleep(FILE_WATCH_INTERVAL_MS)
-      record._sleep = sh
-      pcall(function() sh:await() end)
-      record._sleep = nil
-      if record.cancelled or not server_is_live(sid) then break end
 
-      local cur = scan_tree(base, matches)
+-- The event URI for one change, as `finish_group_scan` emits it. A
-      -- The seam that makes the recheck below WITNESSABLE. `scan_tree`
+-- named function rather than an inline concat so the root-boundary
-      -- suspends on `read_dir` once per directory, and the race is a
+-- witness can drive the PRODUCTION construction at base "/" — a base
-      -- cancel arriving during one of those suspensions --- which no
+-- no fixture can walk for real.
-      -- arrangement of real timing can be made to happen on demand.
+local function watch_change_uri(base, rel)
-      -- Same reason `git.lua` exposes `_deliver_status`: the contract is
+  return file_uri_for(join_under(base, rel))
-      -- about an interleaving the caller does not choose. Unset in
+end
-      -- production, so this costs one nil test per tick.
+
-      -- `cur` is handed over so a test can cancel on THE SCAN THAT
+-- Exposed for the root-boundary witness (the `_deliver_status`
-      -- OBSERVED a given change. Cancelling on any other scan is not a
+-- pattern): the exact functions the watcher matches subjects and
-      -- witness: the loop would break at the post-sleep check on the
+-- builds URIs with. Test-only by convention; production never reads
-      -- next iteration and emit nothing anyway, so the assertion would
+-- them back.
-      -- pass with the recheck below deleted.
+pmacs.lsp._watch_matcher_for_tests = make_matcher
-      if pmacs.lsp._after_scan_for_tests then
+pmacs.lsp._watch_change_uri_for_tests = watch_change_uri
-        pcall(pmacs.lsp._after_scan_for_tests, record, cur)
+
+-- scan_groups[skey .. "\0" .. base] = one scan group per
+-- (server, base): the shared snapshot, the members it serves, and the
+-- D3 state machine — single-flight (`in_flight`), completion-advanced
+-- deadlines (`next_scan_at`), a scan `generation` that rejects stale
+-- completions, `rescan_queued` for joins landing mid-walk, the
+-- backoff `interval`, and the `failure_reported` dedup latch.
+local scan_groups = {}
+
+local function group_key(skey, base)
+  return skey .. "\0" .. base
+end
+
+-- Retire a group: forget it and cancel any in-flight walk
+-- cooperatively. The walk's completion is then rejected by the
+-- identity/generation check in `finish_group_scan`, so a retired
+-- group's state can never be written again.
+local function retire_group(group)
+  scan_groups[group.key] = nil
+  if group.in_flight and group.in_flight.handle then
+    pcall(function()
+      group.in_flight.handle:cancel()
+    end)
+  end
+end
+
+-- One scan's completion. `scan_members` is the membership captured at
+-- scan START — a watcher joining mid-walk is not in it and waits for
+-- its queued baseline scan. Three disjoint arms (framing round 3):
+-- stale/retired, live non-success, success.
+local function finish_group_scan(group, gen, scan_members, ok, result)
+  -- Stale/retired arm: the group was retired (or superseded under the
+  -- same key) while the walk was parked. Nothing here may touch a
+  -- successor's state — #234's P2 recheck, applied at group scope.
+  if scan_groups[group.key] ~= group then return end
+  if not group.in_flight or group.in_flight.generation ~= gen then return end
+  group.in_flight = nil
+  local now = pmacs.editor.monotonic_ms()
+
+  if not ok then
+    -- Live non-success arm: no snapshot, no epoch, no emit; the prior
+    -- snapshot and backoff interval survive. A queued join still gets
+    -- its immediate baseline attempt; otherwise reschedule normally.
+    -- `Handle:await()` raises { tag = "cancelled" } (R45) for the
+    -- intentional `workers.cancel-at-point` outcome, which stays
+    -- quiet; a failure is reported once per distinct message until a
+    -- success clears the latch.
+    local tag = type(result) == "table" and result.tag or nil
+    if tag ~= "cancelled" then
+      local msg = type(result) == "table"
+          and tostring(result.message or result.tag)
+        or tostring(result)
+      if group.failure_reported ~= msg then
+        group.failure_reported = msg
+        local report = "lsp: file watch scan failed for " .. group.base .. ": " .. msg
+        if pmacs.error then pcall(pmacs.error, report) end
+        pcall(pmacs.editor.set_status, report)
       end
-      -- RECHECKED AFTER THE SCAN, not only after the sleep (review P2).
+    end
-      -- The coroutine is suspended for most of a tick with `_sleep`
+    if group.rescan_queued then
-      -- already cleared, so a cancel landing there sets `cancelled` and
+      group.rescan_queued = false
-      -- has no sleep to interrupt. Without this line the resumed scan
+      group.next_scan_at = now
-      -- runs on to `did_change_watched_files` below and a SUPERSEDED
+    else
-      -- watcher emits one last batch under its OLD pattern. One batch is
+      group.next_scan_at = now + group.interval
-      -- enough: it is a wrong-pattern notification the server acts on.
+    end
-      if record.cancelled or not server_is_live(sid) then break end
+    return
-      local changes = {}
+  end
-      for rel, sig in pairs(cur) do
+
-        local was = prev[rel]
+  -- Success arm.
-        if was == nil then
+  group.failure_reported = nil
-          if kind_has(kind_mask, 1) then
+  local cur = {}
-            changes[#changes + 1] =
+  for _, e in ipairs(result) do
-              { uri = file_uri_for(base .. "/" .. rel), type = FC_CREATED }
+    if e.kind ~= "dir" then
+      cur[e.name] = table.concat({
+        tostring(e.size), tostring(e.mtime),
+        tostring(e.mtime_nsec), tostring(e.kind),
+      }, "|")
+    end
+  end
+
+  -- The seam that makes cancel-during-scan WITNESSABLE (#234's P2
+  -- device, per member): the race is a cancel landing while the walk
+  -- job is out, which no arrangement of real timing produces on
+  -- demand. Unset in production. Handed the snapshot so a test can
+  -- cancel on THE SCAN THAT OBSERVED a given change; the delivery
+  -- loop below rechecks `cancelled` per member.
+  if pmacs.lsp._after_scan_for_tests then
+    for _, m in ipairs(scan_members) do
+      pcall(pmacs.lsp._after_scan_for_tests, m, cur)
+    end
+  end
+
+  local changes = {}
+  local prev = group.snapshot
+  if prev then
+    -- Diff once; route per member. `bit` is the WatchKind the event
+    -- needs (Create=1, Change=2, Delete=4).
+    local diff = {}
+    for rel, sig in pairs(cur) do
+      local was = prev[rel]
+      if was == nil then
+        diff[#diff + 1] = { rel = rel, type = FC_CREATED, bit = 1 }
+      elseif was ~= sig then
+        diff[#diff + 1] = { rel = rel, type = FC_CHANGED, bit = 2 }
+      end
+    end
+    for rel in pairs(prev) do
+      if cur[rel] == nil then
+        diff[#diff + 1] = { rel = rel, type = FC_DELETED, bit = 4 }
+      end
+    end
+    -- A member delivers only with a baseline (the registration-epoch
+    -- rule: no events for files that predate the join) and only while
+    -- uncancelled. Changes are deduped by (path, type) into the
+    -- server's single notification.
+    local emitted = {}
+    for _, m in ipairs(scan_members) do
+      if not m.cancelled and m.baseline_epoch ~= nil then
+        for _, d in ipairs(diff) do
+          if kind_has(m.kind_mask, d.bit) and m.match_subject(d.rel) then
+            local key = d.type .. " " .. d.rel
+            if not emitted[key] then
+              emitted[key] = true
+              changes[#changes + 1] = {
+                uri = watch_change_uri(group.base, d.rel),
+                type = d.type,
+              }
+            end
           end
-        elseif was ~= sig and kind_has(kind_mask, 2) then
-          changes[#changes + 1] =
-            { uri = file_uri_for(base .. "/" .. rel), type = FC_CHANGED }
         end
       end
-      for rel in pairs(prev) do
+    end
-        if cur[rel] == nil and kind_has(kind_mask, 4) then
+  end
-          changes[#changes + 1] =
+
-            { uri = file_uri_for(base .. "/" .. rel), type = FC_DELETED }
+  group.snapshot = cur
-        end
+  group.epoch = group.epoch + 1
+  -- Baseline assignment: exactly the members captured at scan start
+  -- with no baseline yet — for them this walk is the first one that
+  -- STARTED after their join.
+  for _, m in ipairs(scan_members) do
+    if not m.cancelled and m.baseline_epoch == nil then
+      m.baseline_epoch = group.epoch
+    end
+  end
+
+  if #changes > 0 then
+    pcall(pmacs.lsp.did_change_watched_files, group.sid, changes)
+    group.interval = FILE_WATCH_INTERVAL_MS
+  else
+    -- A quiet scan backs off toward the cap; waiting costs nothing
+    -- under the after-tick cadence, so this bounds scan frequency,
+    -- not sleep-job length (there are no sleeps).
+    group.interval = math.min(group.interval * 2, FILE_WATCH_BACKOFF_CAP_MS)
+  end
+
+  -- Sweep members cancelled outside cancel_watch_records (the seam,
+  -- or a supersede that raced the walk); an empty group retires.
+  local live = {}
+  for _, m in ipairs(group.members) do
+    if not m.cancelled then live[#live + 1] = m end
+  end
+  group.members = live
+  if #group.members == 0 then
+    retire_group(group)
+    return
+  end
+
+  if group.rescan_queued then
+    group.rescan_queued = false
+    group.next_scan_at = now
+  else
+    group.next_scan_at = now + group.interval
+  end
+end
+
+-- Start one walk for `group`. Single-flight is the caller's contract
+-- (the tick skips in-flight groups; joins queue instead) — this only
+-- stamps the generation and captures the delivery membership.
+local function start_group_scan(group)
+  group.generation = group.generation + 1
+  local gen = group.generation
+  local scan_members = {}
+  for i, m in ipairs(group.members) do
+    scan_members[i] = m
+  end
+  local handle = pmacs.fs.walk_tree(group.base)
+  group.in_flight = {
+    generation = gen,
+    started_at = pmacs.editor.monotonic_ms(),
+    handle = handle,
+  }
+  pmacs.async(function()
+    local ok, result = pcall(function()
+      return handle:await()
+    end)
+    finish_group_scan(group, gen, scan_members, ok, result)
+  end)
+end
+
+-- The cadence (D3 framing; autosave's Q#AS2 idiom). One after-tick
+-- subscription drives every group: a clock read and a compare per
+-- frame, no job and no pool thread while waiting. Installed once and
+-- guarded rather than removed, because `pmacs.hook.remove` does not
+-- exist (the P3 gap).
+local watch_tick_installed = false
+local function ensure_watch_tick()
+  if watch_tick_installed then return end
+  watch_tick_installed = true
+  pmacs.hook.add("process.after-tick", function()
+    if next(scan_groups) == nil then return end
+    local now = pmacs.editor.monotonic_ms()
+    for _, group in pairs(scan_groups) do
+      if not server_is_live(group.sid) then
+        retire_group(group)
+      elseif not group.in_flight and now >= group.next_scan_at then
+        start_group_scan(group)
       end
-      if #changes > 0 then
-        pcall(pmacs.lsp.did_change_watched_files, sid, changes)
-      end
-      prev = cur
     end
   end)
 end
 
+-- Join `record` to its (server, base) group, creating the group on
+-- first use. Joins WAKE the group (framing round 2): the deadline
+-- pulls to now, or — mid-walk — exactly one immediate follow-up scan
+-- is queued, so a joiner's baseline is at most the current walk's
+-- remainder plus one walk away, never a backoff cap. The
+-- join-triggered scan does not reset the backoff curve; only observed
+-- changes do.
+local function join_group(sid, skey, base, record)
+  ensure_watch_tick()
+  local key = group_key(skey, base)
+  local group = scan_groups[key]
+  if not group then
+    group = {
+      key = key,
+      sid = sid,
+      skey = skey,
+      base = base,
+      members = {},
+      snapshot = nil,
+      epoch = 0,
+      generation = 0,
+      in_flight = nil,
+      rescan_queued = false,
+      next_scan_at = 0,
+      interval = FILE_WATCH_INTERVAL_MS,
+      failure_reported = nil,
+    }
+    scan_groups[key] = group
+  end
+  group.members[#group.members + 1] = record
+  record.group = group
+  if group.in_flight then
+    group.rescan_queued = true
+  else
+    group.next_scan_at = 0
+  end
+end
+
+-- The workspace directory the registering server actually serves
+-- (Q#D3-3): its spec `root_uri` — verbatim, nil when the server never
+-- asked for a root — then its `cwd`. Configured roots and custom
+-- resolvers (texlab's Q#LX2) are already folded into the spec, which
+-- is why this must NOT be a fresh `pmacs.project.detect`.
+local function server_workspace_dir(sid)
+  local skey = tostring(sid)
+  local ok, rows = pcall(pmacs.lsp.list)
+  if not ok then return nil end
+  for _, row in ipairs(rows) do
+    if tostring(row.id) == skey then
+      if row.root_uri then
+        local p = pmacs.lsp.path_for_uri(row.root_uri)
+        if p then return p end
+      end
+      if row.cwd then return row.cwd end
+      return nil
+    end
+  end
+  return nil
+end
+
 -- Resolve a GlobPattern (string | { baseUri, pattern }) to
 -- (base_dir, pattern, form). The form must travel with the pair: a
 -- RelativePattern's pattern is relative to its baseUri, and dropping
@@ -2152,23 +2383,52 @@ local function resolve_watcher(sid, gp)
     return pmacs.lsp.path_for_uri(gp.baseUri), gp.pattern or "**", "relative"
   end
   if type(gp) == "string" then
+    local form = (gp:sub(1, 1) == "/") and "absolute" or "relative"
+    -- Q#D3-3: the server's own workspace first. The
+    -- attachment-directory guess remains only for a server with
+    -- neither root_uri nor cwd — and even there it must be
+    -- DETERMINISTIC (review blocker): `pairs` order is hash order, so
+    -- the fallback takes the lexicographically smallest attachment
+    -- directory rather than whichever record iteration yields first.
+    -- REACHABLE, not defensive (review round 2 corrected round 1's
+    -- "unreachable" claim here): `pmacs.lsp.spawn` accepts a spec
+    -- with neither field, and for a MARKERLESS file `ensure_server`
+    -- adopts such a live server because its `root_uri` and the
+    -- attach's `key_uri` are both nil — after which any number of
+    -- buffers in different directories can attach to it.
+    local dir = server_workspace_dir(sid)
+    if dir then return dir, gp, form end
+    local fallback = nil
     for _, rec in pairs(attachments) do
       if rec.server == sid and rec.uri then
         local p = pmacs.lsp.path_for_uri(rec.uri)
-        local dir = p and p:match("^(.*)/[^/]*$")
+        local d = p and p:match("^(.*)/[^/]*$")
-        if dir then
+        -- A file at the filesystem root leaves the capture empty.
-          return dir, gp, (gp:sub(1, 1) == "/") and "absolute" or "relative"
+        if d == "" then d = "/" end
-        end
+        if d and (fallback == nil or d < fallback) then fallback = d end
       end
     end
+    if fallback then return fallback, gp, form end
   end
   return nil, nil, nil
 end
 
+-- Cancel records and leave their groups; a group losing its last
+-- member retires (its in-flight walk cancelled cooperatively).
 local function cancel_watch_records(recs)
   for _, r in ipairs(recs or {}) do
     r.cancelled = true
-    if r._sleep then pcall(function() r._sleep:cancel() end) end
+    local group = r.group
+    if group then
+      r.group = nil
+      for i, m in ipairs(group.members) do
+        if m == r then
+          table.remove(group.members, i)
+          break
+        end
+      end
+      if #group.members == 0 then retire_group(group) end
+    end
   end
 end
 
@@ -2177,20 +2437,29 @@ local function register_file_watchers(sid, registrations)
   file_watchers[skey] = file_watchers[skey] or {}
   for _, reg in ipairs(registrations or {}) do
     if reg.method == "workspace/didChangeWatchedFiles" then
-      -- Re-registering a live id supersedes it (rust-analyzer does
+      local outgoing = file_watchers[skey][reg.id]
-      -- this): cancel the outgoing records first, because the table
-      -- write below drops the only reference to them and an
-      -- uncancelled record polls until the server dies.
-      cancel_watch_records(file_watchers[skey][reg.id])
       local recs = {}
       for _, w in ipairs((reg.registerOptions or {}).watchers or {}) do
         local base, pat, form = resolve_watcher(sid, w.globPattern)
         if base and pat then
-          local r = { cancelled = false, form = form }
+          local r = {
+            cancelled = false,
+            form = form,
+            kind_mask = w.kind or 7,
+            match_subject = make_matcher(base, pat, form),
+            baseline_epoch = nil,
+          }
           recs[#recs + 1] = r
-          start_file_watcher(sid, base, pat, w.kind or 7, r)
+          join_group(sid, skey, base, r)
         end
       end
+      -- Re-registering a live id supersedes it (rust-analyzer does
+      -- this, #234's D2). The successors joined FIRST, so a same-base
+      -- group stays alive across the hand-over — its in-flight walk
+      -- is not torn down, and the joiners already queued their
+      -- baseline. An id whose successors watch a different base still
+      -- retires the old group when its last member leaves here.
+      cancel_watch_records(outgoing)
       file_watchers[skey][reg.id] = recs
     end
   end
@@ -2225,7 +2494,7 @@ end
 --     family for every attached document so the matching store
 --     (`pmacs.inlay_hint` / `pmacs.semantic_tokens`) stays fresh.
 --   * `client/registerCapability` / `client/unregisterCapability` —
---     start/stop the file-watch coroutines for any
+--     join/leave the file-watch scan groups for any
 --     `workspace/didChangeWatchedFiles` registration; reply `null`.
 --
 -- Only servers in `attachments` are drained, so a test (or package)

docs/active-work.md

@@ -250,9 +250,58 @@ the same day (`b867f64`), refreshed and re-gated on the merged base.
 **D3 — the polling cost — is the remainder, and the user has ruled it
 is next (2026-08-11).** **Branch `lsp-file-watch-d3`** (base
 `githubsucks/main` @ `add0ba1`; the remote ref is authoritative), with
-**framing `docs/lsp-file-watch-d3-framing.md`, revision 4, DRAFT —
+**framing `docs/lsp-file-watch-d3-framing.md`, revision 4, APPROVED
-review corrections absorbed; awaiting the four user rulings**, committed
+2026-08-11 with the four rulings adopted as proposed** (honest ⋯N bar;
-at the branch's first commit so it is portable during review. **Review
+no exclusions; server root then cwd then attachment fallback;
+constants). **IMPLEMENTED on the branch**: `pmacs.fs.walk_tree` (one
+cancellable job per scan, eight Rust unit tests), the group scheduler
+in `lsp.lua` (after-tick cadence, single-flight state machine with
+the round-3 non-success partition, registration epochs, backoff,
+retirement), and eighteen acceptance tests — the six #234 tests
+byte-unchanged plus twelve witnesses, each mutation-verified. Two
+implementation-time facts worth keeping:
+
+- **Retirement is deliberately double-enforced** (the unregister path
+  and the post-scan sweep), and the mutation pass proved it: biting
+  either copy alone is masked by the other; only biting both goes red.
+  The sweep exists for seam-cancelled members, the unregister path for
+  idle groups whose next scan may be seconds away.
+- **A second pre-commit round found three more** (implementation
+  review, not framing): mid-walk cancellation was unwitnessed — both
+  Rust cancel tests pre-cancelled and the acceptance test cancelled a
+  QUEUED walk, so deleting the internal polls left everything green
+  (a `cfg(test)` entry hook now cancels at an exact entry boundary
+  and asserts the walk stopped NEAR it, and the retirement witness
+  holds a walk in flight across the unregister and asserts the job
+  settles cancelled); the "unreachable fallback" claim was WRONG — a
+  manual `pmacs.lsp.spawn` may omit both `cwd` and `root_uri`, and
+  `ensure_server` adopts such a server for markerless files (nil ==
+  nil), so the deterministic minimum now has a five-directory
+  through-the-server witness (five, because with two the build's hash
+  order coincided with the lexicographic answer and the first-pairs
+  bite survived); and the root-boundary joins gained a witness
+  through the exported production matcher/URI functions, since no
+  fixture can walk `/` for real.
+- **A pre-commit review round found four blockers**, fixed before
+  anything was committed: empty-tree cancellation (the entry loops
+  never run, so a pre-cancelled walk returned empty SUCCESS — the
+  deletion-storm shape the non-success arm exists to prevent); the
+  defensive attachment fallback was still `pairs`-order
+  nondeterministic (now lexicographic-minimum, with its
+  unreachability-through-production-spawning recorded at the site); a
+  filesystem-root base joined as `//path` (now `join_under`, dired's
+  idiom); and two test probes defaulted on error, so a broken pair
+  could compare equal and lie green (every probe now `expect`s).
+- **A stray marker high in the tree re-roots every markerless fixture
+  under it.** An empty `/tmp/.git` (leftover from the #233
+  investigation, since removed by the user) made project detection
+  root tempdir fixtures at `/tmp`, which under Q#D3-3 the watcher then
+  faithfully watched. Any machine can grow one; a markerless-fixture
+  red that looks like a watcher bug may be an ancestor marker.
+
+Framing and lane were committed
+at the branch's first commit so the document stayed portable during
+review. **Review
 round 3 (2026-08-11) found the live group's non-success transition
 missing**: every job is user-cancellable and `Handle:await()` raises on
 cancel/failure, so an uncaught result could leave `in_flight` set forever.
@@ -285,10 +334,10 @@ became **none** (any unconditional skip deviates from the registered
 glob contract, and `walk_tree` removes the job-count economics that
 motivated it), and the cost arithmetic was corrected to this
 checkout (1,326 jobs/tick for rust-analyzer's six watchers; revised
-steady state is zero jobs at idle). Four open rulings (Q#D3-1..4:
+steady state is zero jobs at idle). The four rulings (Q#D3-1..4)
-the acceptance bar, exclusions, the scan root, knobs vs constants)
+were ADOPTED AS PROPOSED at the 2026-08-11 approval and are recorded
-block implementation. What was known before framing, verified while
+in the framing's status block. What was known before framing,
-framing D1/D2:
+verified while framing D1/D2:
 
 - After #234 the watcher is *correct* but still walks: `walk` recurses
   unconditionally and `matches` gates only recording, so rust-analyzer

docs/lsp-file-watch-d3-framing.md

@@ -1,8 +1,12 @@
 # LSP file watcher D3 — the polling cost — framing
 
-**Status: revision 4 — DRAFT, review corrections absorbed; awaiting
+**Status: revision 4 — APPROVED 2026-08-11.** The user's own review
-the user rulings Q#D3-1..4. No implementation may begin from this
+pass (round 3, absorbed below) closed the state machine; on the
-document.**
+soundness confirmation the four rulings were adopted as proposed:
+**Q#D3-1** the honest bar — absent at idle, one attributable job per
+concurrently due group; **Q#D3-2** no exclusions by default;
+**Q#D3-3** server `root_uri` → `cwd` → attachment fallback;
+**Q#D3-4** constants, no config keys.
 
 Continues issue #233, which stays open until this lane closes it. D1
 and D2 — matching correctness and the re-registration leak — merged as
@@ -374,7 +378,7 @@ jobs at idle**, with one `walk_tree` job per group for the few
 milliseconds each scan actually runs — at most every 250 ms under
 activity and every 4 s at rest, immediately once at registration.
 
-## Open rulings — each blocks implementation
+## The rulings — adopted as proposed at approval (2026-08-11)
 
 - **Q#D3-1 — the acceptance bar, stated accurately (round 2).** At
   idle the indicator is **absent** (no running job exists —
@@ -382,22 +386,23 @@ activity and every 4 s at rest, immediately once at registration.
   shows **one attributable job per concurrently due group** — `⋯N`
   when N (server, base) groups are due on the same frame, each named
   for its root; a typical single-project session has one group.
-  Alternative if `⋯1` must be guaranteed: a global scan queue
+  **Adopted.** The alternative — a global scan queue guaranteeing
-  serializing walks across groups, at the cost of coupling one
+  `⋯1` at the cost of coupling one server's scan latency to
-  server's scan latency to another's tree size. Which bar?
+  another's tree size — was declined.
-- **Q#D3-2 — exclusions.** Proposed: none by default, with opt-in
+- **Q#D3-2 — exclusions.** **Adopted: none by default**, with opt-in
-  exclusion as a documented contract trade (option b) if a user
+  exclusion as a documented contract trade (option b) if a user ever
-  asks. Confirm, or rule for one of (b)/(c)/(d) above.
+  asks.
-- **Q#D3-3 — the scan root.** Proposed: server `root_uri` → server
+- **Q#D3-3 — the scan root.** **Adopted: server `root_uri` → server
-  `cwd` → attached-file directory. This widens the watched tree for
+  `cwd` → attached-file directory** (the fallback itself made
-  servers with a real root (today it is one attached file's
+  deterministic in implementation review: lexicographically smallest
-  directory, chosen by hash order) — a behavioural change to a path
+  attachment directory). This widens the watched tree for servers
-  real servers exercise. Confirm the order, or rule otherwise.
+  with a real root — a behavioural change to a path real servers
-- **Q#D3-4 — interval, cap, and backoff curve: constants or config
+  exercise, accepted as such.
-  keys.** The D1/D2 framing refused a knob for a defect; with D3 the
+- **Q#D3-4 — interval, cap, and backoff curve.** **Adopted:
-  cadence becomes a designed mechanism, so keys are defensible — but
+  constants until someone asks.** The D1/D2 framing refused a knob
-  more registry surface is coherence cost. Proposed: constants until
+  for a defect; with D3 the cadence is a designed mechanism, so keys
-  someone asks.
+  would be defensible — but more registry surface is coherence cost
+  nobody has yet paid for a reason.
 
 ## Verification sketch
 

src/async_runtime.rs

@@ -317,6 +317,10 @@ pub enum JobKind {
     Parse,
     /// `dispatch_fs_read_dir` --- directory enumeration ([T M8.1]).
     FsReadDir,
+    /// `dispatch_fs_walk_tree` --- one whole-tree enumeration as a
+    /// single job (issue #233 D3: the watcher's per-directory job
+    /// storm collapsed into one purpose per scan).
+    FsWalkTree,
     /// `dispatch_fs_stat` --- single-path metadata ([T M8.1]).
     FsStat,
     /// `dispatch_fs_rename` --- atomic rename ([T M8.1]).
@@ -355,6 +359,7 @@ impl JobKind {
             JobKind::Grep => "grep",
             JobKind::Parse => "parse",
             JobKind::FsReadDir => "fs_read_dir",
+            JobKind::FsWalkTree => "fs_walk_tree",
             JobKind::FsStat => "fs_stat",
             JobKind::FsRename => "fs_rename",
             JobKind::FsChmod => "fs_chmod",
@@ -1185,6 +1190,29 @@ impl AsyncRuntime {
         id
     }
 
+    /// Dispatch a `walk_tree(base)` job: the whole recursive tree as
+    /// ONE job, entry names base-relative (issue #233 D3). The reply
+    /// reuses [`ReplyKind::ReadDir`] --- the payload shape is
+    /// identical and the Lua boundary needs no second conversion;
+    /// [`JobKind::FsWalkTree`] still separates the two for the
+    /// `*workers*` label and the purpose string. Cancellation and
+    /// tolerance are [`crate::fs::walk_tree_blocking`]'s contract.
+    pub fn dispatch_fs_walk_tree(&self, base: PathBuf, supersede: Option<&str>) -> JobId {
+        let (id, cancel) = self.allocate(JobSpec {
+            kind: JobKind::FsWalkTree,
+            supersede,
+            stream: None,
+            resource: None,
+            purpose: format!("walk_tree {}", base.display()),
+        });
+        let bus = self.workers.clone();
+        self.pool.dispatch(move |_pool| {
+            let kind = run_fs_walk_tree(&cancel, &base);
+            let _ = bus.send(ASYNC_REPLY_TOPIC, &WorkerReply { job_id: id, kind });
+        });
+        id
+    }
+
     /// Dispatch a `stat(path)` job. Returns one [`FsDirEntry`] of
     /// metadata for `path`. T M8.1.
     pub fn dispatch_fs_stat(&self, path: PathBuf, supersede: Option<&str>) -> JobId {
@@ -1759,6 +1787,16 @@ fn run_fs_read_dir(
     }
 }
 
+fn run_fs_walk_tree(cancel: &CancellationToken, base: &Path) -> ReplyKind {
+    match crate::fs::walk_tree_blocking(base, cancel) {
+        Ok(listing) => ReplyKind::ReadDir(listing),
+        Err(FsError::Cancelled) => ReplyKind::Cancelled,
+        Err(e @ (FsError::Io { .. } | FsError::NonUtf8Path { .. })) => {
+            ReplyKind::Error(e.to_string())
+        }
+    }
+}
+
 fn run_fs_stat(cancel: &CancellationToken, path: &Path) -> ReplyKind {
     match stat_blocking(path, cancel) {
         Ok(entry) => ReplyKind::Stat(entry),

src/bin/pmacs_fake_lsp.rs

@@ -88,6 +88,8 @@ fn main() {
     let mut stdout = io::stdout().lock();
     let mut crashed_after_init = false;
     let mut open_docs: HashMap<String, String> = HashMap::new();
+    // `filewatchjoin` / `filewatchretire` mid-session triggers.
+    let mut didchange_count: u32 = 0;
     // `fullonly` observability: counts /full responses (rid-1, rid-2…).
     let mut full_count: u32 = 0;
     loop {
@@ -354,6 +356,58 @@ fn main() {
                 });
                 write_frame(&mut stdout, &req);
             }
+            // Issue #233 D3: `filewatchjoin` registers one watcher at
+            // initialized and a SECOND one — same base, different
+            // pattern — on the first didChange it receives. The
+            // didChange is the test's trigger for a mid-session join,
+            // which is what the join-wakes / registration-epoch /
+            // queued-baseline witnesses need and no
+            // at-initialized-only mode can produce.
+            ("initialized", _) if mode == "filewatchjoin" => {
+                let base = std::env::var("PMACS_FAKE_LSP_WATCH_BASE").unwrap_or_default();
+                let req = serde_json::json!({
+                    "jsonrpc": "2.0",
+                    "id": 9310,
+                    "method": "client/registerCapability",
+                    "params": { "registrations": [{
+                        "id": "watch-j1",
+                        "method": "workspace/didChangeWatchedFiles",
+                        "registerOptions": { "watchers": [{
+                            "globPattern": {
+                                "baseUri": format!("file://{base}"),
+                                "pattern": "**/*.aaa"
+                            },
+                            "kind": 7
+                        }] }
+                    }] }
+                });
+                write_frame(&mut stdout, &req);
+            }
+            // Issue #233 D3: `filewatchretire` registers at
+            // initialized, UNREGISTERS on the first didChange (the
+            // group's last member leaves — retirement), and
+            // re-registers on the second (a fresh group whose
+            // baseline folds whatever happened while unwatched).
+            ("initialized", _) if mode == "filewatchretire" => {
+                let base = std::env::var("PMACS_FAKE_LSP_WATCH_BASE").unwrap_or_default();
+                let req = serde_json::json!({
+                    "jsonrpc": "2.0",
+                    "id": 9320,
+                    "method": "client/registerCapability",
+                    "params": { "registrations": [{
+                        "id": "watch-r1",
+                        "method": "workspace/didChangeWatchedFiles",
+                        "registerOptions": { "watchers": [{
+                            "globPattern": {
+                                "baseUri": format!("file://{base}"),
+                                "pattern": "**/*.txt"
+                            },
+                            "kind": 7
+                        }] }
+                    }] }
+                });
+                write_frame(&mut stdout, &req);
+            }
             // Issue #233 review P1 guard: `filewatchbare` registers a
             // BARE STRING with no base and no leading `/` — `*.txt`.
             // The string arm and the `filewatchflat` arm below carry the
@@ -563,6 +617,63 @@ fn main() {
                 }
             }
             ("textDocument/didOpen" | "textDocument/didChange", _) => {
+                if method == "textDocument/didChange" {
+                    didchange_count += 1;
+                    let base = std::env::var("PMACS_FAKE_LSP_WATCH_BASE").unwrap_or_default();
+                    // `filewatchjoin` / `filewatchretire`: a didChange
+                    // is the test's mid-session trigger; see the
+                    // `initialized` arms above.
+                    if mode == "filewatchjoin" && didchange_count == 1 {
+                        let req = serde_json::json!({
+                            "jsonrpc": "2.0",
+                            "id": 9311,
+                            "method": "client/registerCapability",
+                            "params": { "registrations": [{
+                                "id": "watch-j2",
+                                "method": "workspace/didChangeWatchedFiles",
+                                "registerOptions": { "watchers": [{
+                                    "globPattern": {
+                                        "baseUri": format!("file://{base}"),
+                                        "pattern": "**/*.bbb"
+                                    },
+                                    "kind": 7
+                                }] }
+                            }] }
+                        });
+                        write_frame(&mut stdout, &req);
+                    }
+                    if mode == "filewatchretire" && didchange_count == 1 {
+                        let req = serde_json::json!({
+                            "jsonrpc": "2.0",
+                            "id": 9321,
+                            "method": "client/unregisterCapability",
+                            "params": { "unregisterations": [{
+                                "id": "watch-r1",
+                                "method": "workspace/didChangeWatchedFiles"
+                            }] }
+                        });
+                        write_frame(&mut stdout, &req);
+                    }
+                    if mode == "filewatchretire" && didchange_count == 2 {
+                        let req = serde_json::json!({
+                            "jsonrpc": "2.0",
+                            "id": 9322,
+                            "method": "client/registerCapability",
+                            "params": { "registrations": [{
+                                "id": "watch-r2",
+                                "method": "workspace/didChangeWatchedFiles",
+                                "registerOptions": { "watchers": [{
+                                    "globPattern": {
+                                        "baseUri": format!("file://{base}"),
+                                        "pattern": "**/*.txt"
+                                    },
+                                    "kind": 7
+                                }] }
+                            }] }
+                        });
+                        write_frame(&mut stdout, &req);
+                    }
+                }
                 let uri = params
                     .get("textDocument")
                     .and_then(|t| t.get("uri"))

src/fs.rs

@@ -377,6 +377,144 @@ pub fn read_dir_blocking(
     })
 }
 
+/// Recursively walk `base` into ONE flat listing whose entry names are
+/// base-RELATIVE paths (`sub/dir/file.txt`). Issue #233 D3: the LSP
+/// file-watcher scan used to be one `read_dir` job per directory per
+/// tick; this is the whole tree as one job.
+///
+/// Contract, in the D3 framing's terms:
+/// - **Symlinks are recorded, never traversed.** `lstat` kind
+///   `Symlink` regardless of target, so a link cycle cannot loop the
+///   walk. A dangling or unreadable link target leaves the entry in
+///   the listing with `symlink_target = None`, the same tolerance
+///   [`read_dir_blocking`] applies.
+/// - **Cancellation is cooperative and prompt**: the token is polled
+///   once per directory and every [`READDIR_CANCEL_POLL_EVERY`]
+///   entries within one --- the cadence [`read_dir_blocking`] set.
+/// - **The root failing to open fails the walk** (the caller's
+///   live-failure arm owns it); an unreadable SUBDIRECTORY is skipped
+///   with its whole subtree, which is the Lua `scan_tree` `pcall`
+///   behaviour this primitive replaces. A non-UTF-8 entry name is
+///   skipped the same way: such a path cannot become a watcher URI,
+///   and one weird name must not take the walk down.
+///
+/// Directory entries appear in the listing (kind `dir`) so a consumer
+/// can see structure; the watcher filters them out when building
+/// signatures, as its Lua walk always did.
+pub fn walk_tree_blocking(
+    base: &Path,
+    cancel: &CancellationToken,
+) -> Result<FsDirListing, FsError> {
+    // Checked BEFORE opening and again before returning, not only
+    // inside the entry loops: an empty tree never enters a loop, so a
+    // pre-cancelled queued walk would otherwise return an empty
+    // SUCCESS — which the scheduler's success arm would commit as a
+    // snapshot and diff into a deletion storm, the exact outcome the
+    // live non-success arm exists to prevent. (And a missing root
+    // must report Cancelled, not its Io error, for the same reason.)
+    if cancel.is_cancelled() {
+        return Err(FsError::Cancelled);
+    }
+    let root = std::fs::read_dir(base).map_err(|source| FsError::Io {
+        path: base.display().to_string(),
+        source,
+    })?;
+    let mut out: Vec<FsDirEntry> = Vec::new();
+    // Subdirectories discovered but not yet walked, with their
+    // base-relative prefixes. LIFO order --- traversal order is not
+    // part of the contract; the consumer diffs a map.
+    let mut pending: Vec<(std::path::PathBuf, String)> = Vec::new();
+    walk_one_dir(root, "", cancel, &mut out, &mut pending)?;
+    while let Some((dir, prefix)) = pending.pop() {
+        if cancel.is_cancelled() {
+            return Err(FsError::Cancelled);
+        }
+        // Subtree skip on an unreadable directory: scan_tree's pcall.
+        let Ok(iter) = std::fs::read_dir(&dir) else {
+            continue;
+        };
+        walk_one_dir(iter, &prefix, cancel, &mut out, &mut pending)?;
+    }
+    if cancel.is_cancelled() {
+        return Err(FsError::Cancelled);
+    }
+    Ok(FsDirListing {
+        entries: out,
+        errors: None,
+    })
+}
+
+#[cfg(test)]
+thread_local! {
+    /// Test-only seam: invoked after every entry [`walk_one_dir`]
+    /// records. A MID-walk cancellation is an interleaving no real
+    /// timing produces on demand (the `_after_scan_for_tests`
+    /// justification, at this layer): the hook lets a test flip the
+    /// cancel token at an exact entry boundary and assert the walk
+    /// stopped NEAR it --- which is what discriminates the internal
+    /// polls from the entry/exit checks alone. `None` outside the one
+    /// test that arms it.
+    static WALK_ENTRY_HOOK: std::cell::RefCell<Option<Box<dyn FnMut()>>> =
+        const { std::cell::RefCell::new(None) };
+}
+
+/// One directory's worth of [`walk_tree_blocking`]: record every
+/// representable entry under its base-relative name and queue child
+/// directories. Only cancellation propagates as an error --- every
+/// per-entry failure is a skip, per the walk's tolerance contract.
+fn walk_one_dir(
+    iter: std::fs::ReadDir,
+    prefix: &str,
+    cancel: &CancellationToken,
+    out: &mut Vec<FsDirEntry>,
+    pending: &mut Vec<(std::path::PathBuf, String)>,
+) -> Result<(), FsError> {
+    for (i, entry_result) in iter.enumerate() {
+        if i % READDIR_CANCEL_POLL_EVERY == 0 && cancel.is_cancelled() {
+            return Err(FsError::Cancelled);
+        }
+        let Ok(entry) = entry_result else { continue };
+        let Some(name) = entry.file_name().to_str().map(ToOwned::to_owned) else {
+            continue;
+        };
+        let rel = if prefix.is_empty() {
+            name
+        } else {
+            format!("{prefix}/{name}")
+        };
+        let entry_path = entry.path();
+        let Ok(metadata) = std::fs::symlink_metadata(&entry_path) else {
+            continue;
+        };
+        let kind = classify(&metadata);
+        if matches!(kind, FsEntryKind::Dir) {
+            pending.push((entry_path.clone(), rel.clone()));
+        }
+        let mut symlink_target = None;
+        if matches!(kind, FsEntryKind::Symlink)
+            && let Ok(target) = std::fs::read_link(&entry_path)
+        {
+            symlink_target = target.to_str().map(ToOwned::to_owned);
+        }
+        out.push(FsDirEntry {
+            name: rel,
+            kind,
+            size: metadata.len(),
+            mtime_secs: mtime_to_unix_secs(&metadata),
+            mtime_nsec: mtime_to_unix_nsec(&metadata),
+            mode: mode_bits(&metadata),
+            symlink_target,
+        });
+        #[cfg(test)]
+        WALK_ENTRY_HOOK.with(|h| {
+            if let Some(hook) = h.borrow_mut().as_mut() {
+                hook();
+            }
+        });
+    }
+    Ok(())
+}
+
 /// Route one per-entry failure: append it to the tolerant channel, or
 /// propagate it when the caller asked for the fatal contract.
 ///
@@ -930,4 +1068,185 @@ mod tests {
             other => panic!("expected NonUtf8Path, got {other:?}"),
         }
     }
+
+    #[test]
+    fn walk_tree_flattens_with_relative_names() {
+        let td = tempfile::tempdir().expect("tempdir");
+        std::fs::write(td.path().join("a.txt"), b"hello").expect("write");
+        std::fs::create_dir(td.path().join("sub")).expect("mkdir");
+        std::fs::write(td.path().join("sub/b.txt"), b"xy").expect("write");
+        std::fs::create_dir(td.path().join("sub/deep")).expect("mkdir");
+        std::fs::write(td.path().join("sub/deep/c.txt"), b"z").expect("write");
+        let listing = walk_tree_blocking(td.path(), &token()).expect("walk");
+        assert!(listing.errors.is_none());
+        let mut names: Vec<&str> = listing.entries.iter().map(|e| e.name.as_str()).collect();
+        names.sort_unstable();
+        assert_eq!(
+            names,
+            vec!["a.txt", "sub", "sub/b.txt", "sub/deep", "sub/deep/c.txt"]
+        );
+        let c = listing
+            .entries
+            .iter()
+            .find(|e| e.name == "sub/deep/c.txt")
+            .unwrap();
+        assert_eq!(c.kind, FsEntryKind::File);
+        assert_eq!(c.size, 1);
+        let sub = listing.entries.iter().find(|e| e.name == "sub").unwrap();
+        assert_eq!(sub.kind, FsEntryKind::Dir);
+    }
+
+    #[test]
+    fn walk_tree_records_symlink_without_traversing() {
+        let td = tempfile::tempdir().expect("tempdir");
+        std::fs::write(td.path().join("real.txt"), b"x").expect("write");
+        // A link back to the root: traversal would loop forever, so
+        // completing at all is half the witness.
+        symlink(td.path(), td.path().join("loop")).expect("symlink");
+        let listing = walk_tree_blocking(td.path(), &token()).expect("walk");
+        let link = listing.entries.iter().find(|e| e.name == "loop").unwrap();
+        assert_eq!(link.kind, FsEntryKind::Symlink);
+        assert!(
+            !listing.entries.iter().any(|e| e.name.starts_with("loop/")),
+            "a symlinked directory must be recorded, never entered"
+        );
+    }
+
+    #[test]
+    fn walk_tree_skips_unreadable_subdirectory_and_keeps_the_rest() {
+        use std::os::unix::fs::PermissionsExt;
+        let td = tempfile::tempdir().expect("tempdir");
+        std::fs::write(td.path().join("a.txt"), b"x").expect("write");
+        std::fs::create_dir(td.path().join("locked")).expect("mkdir");
+        std::fs::write(td.path().join("locked/hidden.txt"), b"x").expect("write");
+        std::fs::set_permissions(
+            td.path().join("locked"),
+            std::fs::Permissions::from_mode(0o000),
+        )
+        .expect("chmod 000");
+        let result = walk_tree_blocking(td.path(), &token());
+        // Restore before asserting so the tempdir can be removed even
+        // if an assertion fails.
+        std::fs::set_permissions(
+            td.path().join("locked"),
+            std::fs::Permissions::from_mode(0o755),
+        )
+        .expect("chmod back");
+        let listing = result.expect("walk must survive an unreadable subdirectory");
+        let names: Vec<&str> = listing.entries.iter().map(|e| e.name.as_str()).collect();
+        assert!(names.contains(&"a.txt"));
+        assert!(
+            names.contains(&"locked"),
+            "the dir entry itself is representable"
+        );
+        assert!(
+            !names.contains(&"locked/hidden.txt"),
+            "the unreadable subtree is skipped, matching scan_tree's pcall"
+        );
+    }
+
+    #[test]
+    fn walk_tree_polls_cancellation_token() {
+        let td = tempfile::tempdir().expect("tempdir");
+        std::fs::write(td.path().join("a.txt"), b"x").expect("write");
+        let cancel = token();
+        cancel.cancel();
+        let err = walk_tree_blocking(td.path(), &cancel).expect_err("must observe cancel");
+        assert!(matches!(err, FsError::Cancelled));
+    }
+
+    /// Review blocker: the entry loops never run on an EMPTY tree, so
+    /// without the entry/exit checks a pre-cancelled queued walk
+    /// returned an empty SUCCESS — which the scheduler would commit as
+    /// a snapshot and diff into a deletion storm. A missing root must
+    /// likewise report Cancelled, not its Io error.
+    #[test]
+    fn walk_tree_pre_cancelled_is_cancelled_even_with_no_entries() {
+        let empty = tempfile::tempdir().expect("tempdir");
+        let cancel = token();
+        cancel.cancel();
+        let err = walk_tree_blocking(empty.path(), &cancel)
+            .expect_err("empty tree must still observe cancel");
+        assert!(matches!(err, FsError::Cancelled));
+
+        let err = walk_tree_blocking(&empty.path().join("absent"), &cancel)
+            .expect_err("missing root must still observe cancel");
+        assert!(
+            matches!(err, FsError::Cancelled),
+            "cancellation outranks the root error: got {err:?}"
+        );
+    }
+
+    /// Review blocker: both prior cancellation tests pre-cancelled, so
+    /// deleting the INTERNAL polls (per directory, and every
+    /// [`READDIR_CANCEL_POLL_EVERY`] entries) left every test green —
+    /// the walk ran to completion and only the exit check fired. This
+    /// cancels MID-walk through the test hook and asserts the walk
+    /// stopped near the cancellation point, not at the end.
+    #[test]
+    fn walk_tree_observes_cancellation_mid_walk() {
+        let td = tempfile::tempdir().expect("tempdir");
+        for d in 0..3 {
+            let dir = td.path().join(format!("d{d}"));
+            std::fs::create_dir(&dir).expect("mkdir");
+            for i in 0..41 {
+                std::fs::write(dir.join(format!("f{i:02}.txt")), b"x").expect("write");
+            }
+        }
+        let cancel = token();
+        let seen = std::rc::Rc::new(std::cell::Cell::new(0usize));
+        {
+            let seen = seen.clone();
+            let cancel = cancel.clone();
+            WALK_ENTRY_HOOK.with(|h| {
+                *h.borrow_mut() = Some(Box::new(move || {
+                    let n = seen.get() + 1;
+                    seen.set(n);
+                    if n == 5 {
+                        cancel.cancel();
+                    }
+                }));
+            });
+        }
+        let result = walk_tree_blocking(td.path(), &cancel);
+        WALK_ENTRY_HOOK.with(|h| *h.borrow_mut() = None);
+        assert!(
+            matches!(result, Err(FsError::Cancelled)),
+            "mid-walk cancel must surface as Cancelled"
+        );
+        // 126 entries total (3 dirs + 123 files). The next entry poll
+        // after the cancel at entry 5 is at most one
+        // READDIR_CANCEL_POLL_EVERY stride away.
+        assert!(
+            seen.get() < 60,
+            "the walk must stop near the mid-walk cancel, not run the \
+             whole tree ({} of 126 entries processed)",
+            seen.get()
+        );
+    }
+
+    #[test]
+    fn walk_tree_fails_when_the_root_cannot_open() {
+        let td = tempfile::tempdir().expect("tempdir");
+        let err = walk_tree_blocking(&td.path().join("absent"), &token())
+            .expect_err("missing root must fail the walk");
+        assert!(matches!(err, FsError::Io { .. }));
+    }
+
+    #[test]
+    fn walk_tree_matches_read_dir_on_a_flat_directory() {
+        let td = tempfile::tempdir().expect("tempdir");
+        std::fs::write(td.path().join("a.txt"), b"hello").expect("write");
+        std::fs::write(td.path().join("b.md"), b"xy").expect("write");
+        let mut walked = walk_tree_blocking(td.path(), &token())
+            .expect("walk")
+            .entries;
+        let mut listed = read_dir_fatal(td.path(), &token()).expect("read_dir");
+        walked.sort_by(|a, b| a.name.cmp(&b.name));
+        listed.sort_by(|a, b| a.name.cmp(&b.name));
+        // At depth zero the relative name IS the basename, so the two
+        // primitives must agree entry-for-entry --- the signature
+        // parity the D3 framing requires.
+        assert_eq!(walked, listed);
+    }
 }

src/lua_bindings/mod.rs

@@ -7364,6 +7364,16 @@ pub fn install_async(
         )?;
     }
 
+    {
+        let rt = runtime.clone();
+        async_mod.set(
+            "_dispatch_fs_walk_tree",
+            lua.create_function(move |_, (base, key): (String, Option<String>)| {
+                Ok(rt.dispatch_fs_walk_tree(std::path::PathBuf::from(base), key.as_deref()))
+            })?,
+        )?;
+    }
+
     {
         let rt = runtime.clone();
         async_mod.set(