Let the caller participate in directory reads

XFF.md

@@ -564,7 +564,7 @@ See also: [Configuration](#topic-config), [Archives](#topic-archive), [Output](#
 <a id="flag-jobs"></a>
 
 - `-j N, -j=N, --jobs=N|all` - directory-read, eligible content-match, comparison and -exec workers _(global, find, xff, rg)_
-  `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories using one command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in eligible content and comparison work. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core.
+  `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories using one command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in directory reads, eligible content and comparison work. It reads an uncached current directory while workers read later siblings, and can claim a required queued read once. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core.
   See also: [Output](#topic-output), [Statistics](#topic-stats)
 
 <a id="flag-sort"></a>

docs/benchmark-correctness-audit.md

@@ -352,6 +352,33 @@ not valid evidence of sufficient remaining work. The bookkeeping estimate, retai
 byte policy and task granularity need allocation profiles and repeated native
 measurements. No timing, total-memory improvement or macOS acceptance is claimed.
 
+### Caller-participation control
+
+A deterministic 65-sibling in-memory control identified a separate gap in the
+bounded-frontier candidate: with `--jobs=2`, only one background reader touched
+child directories, peak concurrent reads were one, and the two-directory
+rendezvous timed out. The caller never read a child directory. With larger
+budgets, the observed reader identities likewise excluded the caller. Ordered
+output, exact-once reads and the thread cap still passed; those properties alone
+did not demonstrate useful caller participation.
+
+The follow-up reserves the first uncached required directory for the caller and
+queues later siblings. A required queued read is represented by move-only work
+that either the caller or worker can claim once. An already-claimed read still
+requires waiting for its result; no arbitrary unrelated job is run on that wait.
+The command budget remains caller-inclusive, with at most `N-1` background
+threads, reused across both comparison inventories and pair comparisons.
+
+Controls verify caller/worker overlap for budgets 2/3/10, serial-equivalent
+ordered visits and exactly one read per child. An occupied-worker control forces
+the caller to read its required queued directory before releasing that worker.
+Executor controls cover queued claims, already-running claims and simultaneous
+caller/worker claims, including move-only producer ownership. Cancellation,
+shared byte/slot accounting, error reporting and borrowed-state lifetime remain
+part of the required validation. The extra claim state and callback allocation
+need native allocation and timing evidence. Remaining-cost admission and useful
+task size are unchanged; no performance or native macOS acceptance is claimed.
+
 ### Memory-accounting finding and remaining acceptance work
 
 Direct `/usr/bin/time -v` runs show approximately 5.3-5.4 MiB peak RSS for a small

docs/design-parallel.md

@@ -159,6 +159,14 @@ The caller claims independent stat, content and comparison chunks rather than
 waiting idle while `N` additional workers run. Sequential comparison inventories
 remove the second coordinator, but may reduce overlap when each tree has a
 narrow frontier on slow storage; native A/B controls must quantify that tradeoff.
+For directory reads, the first uncached required directory stays on the caller,
+while the bounded frontier queues later siblings for background workers. A queued
+read that becomes required can be claimed by its caller or its dispatched worker,
+exactly once. The claim mutex protects callback ownership only; filesystem access
+and result publication happen after unlocking. If a worker already owns that
+required read, the caller waits for its result. It does not execute an unrelated
+slow sibling merely to help the queue. This improves available participation, not
+a guarantee of full utilization or sufficient useful work for each dispatch.
 The bounded-frontier follow-up separates foreground stat/content/comparison jobs
 from speculative directory reads. When both queues contain work, dispatch
 alternates their classes, retaining FIFO order within each class. This is fairness

xff/cli/globals.cc

@@ -907,7 +907,9 @@ constexpr std::array kGlobals = std::to_array<GlobalFlag>({
             "using one command-owned, lazily started pool of up to `N-1` background workers plus the calling "
             "coordinator, reused for directory reads, required eager metadata, eligible content matching "
             "and file comparison. Comparison trees are coordinated sequentially within that same budget. "
-            "The caller participates in eligible content and comparison work. Directory read-ahead uses a bounded "
+            "The caller participates in directory reads, eligible content and comparison work. It reads an uncached "
+            "current directory while workers read later siblings, and can claim a required queued read once. "
+            "Directory read-ahead uses a bounded "
             "ordered window and a shared retained-storage allowance; oversized speculative listings are read on "
             "demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight "
             "filesystem call must finish. Workers reuse the same executor for independent work; small workloads "

xff/engine/BUILD.bazel

@@ -171,6 +171,7 @@ cc_test(
         ":walk_cc",
         "//xff/vfs:local_fs_cc",
         "//xff/vfs:vfs_cc",
+        "@abseil-cpp//absl/base:core_headers",
         "@abseil-cpp//absl/functional:function_ref",
         "@abseil-cpp//absl/status",
         "@abseil-cpp//absl/status:statusor",

xff/engine/command_workers_test.cc

@@ -64,7 +64,7 @@ class CommandFs final : public vfs::FileSystem {
       }
       return entries;
     }
-    Record(path.starts_with("left/") ? 0 : 1, participants_ > 1 ? participants_ - 1 : 1);
+    Record(path.starts_with("left/") ? 0 : 1, participants_);
     return std::vector<vfs::Entry>{
         {
             .path = std::string(path) + "/file",
@@ -133,7 +133,8 @@ class CommandFs final : public vfs::FileSystem {
   void CheckWalkReuse() const {
     const absl::MutexLock lock(mutex_);
     EXPECT_THAT(phase_readers_.at(0), Eq(phase_readers_.at(1)));
-    EXPECT_THAT(phase_readers_.at(0), SizeIs(participants_ > 1 ? participants_ - 1 : 1));
+    EXPECT_THAT(phase_readers_.at(0), SizeIs(participants_));
+    EXPECT_THAT(phase_readers_.at(0), Contains(coordinator_));
   }
 
  private:

xff/engine/run_executor.cc

@@ -22,6 +22,20 @@ struct ReadAheadBudget final {
 };
 }  // namespace executor_detail
 
+RunWork::RunWork(absl::AnyInvocable<void()> work) : work_(std::move(work)) {}
+
+absl::AnyInvocable<void()> RunWork::Take() {
+  const absl::MutexLock lock(mutex_);
+  return std::exchange(work_, absl::AnyInvocable<void()>{});
+}
+
+void RunWork::Run() {
+  auto work = Take();
+  if (work) {
+    work();
+  }
+}
+
 struct ReadAheadReservation::Impl final {
   Impl(std::shared_ptr<executor_detail::ReadAheadBudget> state, std::size_t initial_bytes)
       : budget(std::move(state)), minimum_bytes(initial_bytes), bytes(initial_bytes) {}

xff/engine/run_executor.h

@@ -34,6 +34,30 @@ struct TaskState final {
 
 enum class RunTaskClass { kForeground, kReadAhead };
 
+// A leaf job claimable by its coordinator or a dispatched worker. Exactly one
+// Run call executes the owned job, outside its mutex; other calls return without
+// waiting. Result publication, if needed, remains the job's responsibility.
+// Use shared ownership while both caller and queued worker can access the job.
+class RunWork final {
+ public:
+  explicit RunWork(absl::AnyInvocable<void()> work);
+
+  RunWork(const RunWork&) = delete;
+  RunWork& operator=(const RunWork&) = delete;
+  RunWork(RunWork&&) = delete;
+  RunWork& operator=(RunWork&&) = delete;
+
+  void Run() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+  absl::AnyInvocable<void()> Take() ABSL_LOCKS_EXCLUDED(mutex_);
+
+  // This mutex guards ownership of work_ only and is released before invoking
+  // the job. It is never held with an executor dispatch or read-budget mutex.
+  absl::Mutex mutex_;
+  absl::AnyInvocable<void()> work_ ABSL_GUARDED_BY(mutex_);
+};
+
 struct ReadAheadStats {
   // Unfinished reads. Completed results keep their byte charge, not an execution slot.
   std::size_t in_use = 0;

xff/engine/run_executor_test.cc

@@ -3,6 +3,7 @@
 
 #include "xff/engine/run_executor.h"
 
+#include <atomic>
 #include <limits>
 #include <memory>
 #include <set>
@@ -83,6 +84,112 @@ TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) {
   executor.Submit([] {}).Get();
 }
 
+TEST_F(RunExecutorTest, CallerClaimsRequiredWorkWhileItsQueuedWrapperWaits) {
+  RunExecutor executor(1);
+  executor.Start(1);
+  absl::Notification started;
+  absl::Notification release;
+  auto occupied = executor.Submit([&] {
+    started.Notify();
+    release.WaitForNotification();
+  });
+  const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5));
+  std::size_t calls = 0;
+  RunPromise<int> promise;
+  auto result = promise.Task();
+  const auto work =
+      std::make_shared<RunWork>([&calls, value = std::make_unique<int>(42), promise = std::move(promise)] mutable {
+        ++calls;
+        promise.SetValue(*value);
+      });
+  auto queued = executor.Submit([work] { work->Run(); }, RunTaskClass::kReadAhead);
+  work->Run();
+  const bool ready_before_release = result.Ready();
+  release.Notify();
+  occupied.Get();
+  queued.Get();
+  work->Run();
+  EXPECT_THAT(running, IsTrue());
+  EXPECT_THAT(ready_before_release, IsTrue());
+  EXPECT_THAT(result.Get(), Eq(42));
+  EXPECT_THAT(calls, Eq(1));
+}
+
+TEST_F(RunExecutorTest, CallerDoesNotDuplicateOrWaitInsideWorkerClaimedWork) {
+  RunExecutor executor(1);
+  executor.Start(1);
+  absl::Notification started;
+  absl::Notification release;
+  std::atomic<int> calls = 0;
+  RunPromise<int> promise;
+  auto result = promise.Task();
+  const auto work = std::make_shared<RunWork>([&calls, &started, &release, promise = std::move(promise)] mutable {
+    calls.fetch_add(1);
+    started.Notify();
+    // Bound a broken claim implementation so it reports failure rather than hanging.
+    promise.SetValue(release.WaitForNotificationWithTimeout(absl::Seconds(5)) ? 42 : 0);
+  });
+  auto dispatched = executor.Submit([work] { work->Run(); });
+  const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5));
+  work->Run();
+  const bool ready_before_release = result.Ready();
+  release.Notify();
+  dispatched.Get();
+  EXPECT_THAT(running, IsTrue());
+  EXPECT_THAT(ready_before_release, IsFalse());
+  EXPECT_THAT(result.Get(), Eq(42));
+  EXPECT_THAT(calls.load(), Eq(1));
+}
+
+TEST_F(RunExecutorTest, SimultaneousCallerAndWorkerClaimsExecuteMoveOnlyWorkOnce) {
+  RunExecutor executor(2);
+  executor.Start(2);
+  absl::Barrier ready(3);
+  std::atomic<int> calls = 0;
+  RunPromise<int> promise;
+  auto result = promise.Task();
+  const auto work =
+      std::make_shared<RunWork>([&calls, value = std::make_unique<int>(42), promise = std::move(promise)] mutable {
+        calls.fetch_add(1);
+        promise.SetValue(*value);
+      });
+  const auto claim = [&] {
+    ready.Block();
+    work->Run();
+  };
+  auto first = executor.Submit(claim);
+  auto second = executor.Submit(claim);
+  claim();
+  first.Get();
+  second.Get();
+  EXPECT_THAT(result.Get(), Eq(42));
+  EXPECT_THAT(calls.load(), Eq(1));
+}
+
+TEST_F(RunExecutorTest, WorkerWrapperDoesNotWaitForCallerClaimedWork) {
+  RunExecutor executor(1);
+  executor.Start(1);
+  absl::Notification started;
+  absl::Notification release;
+  RunPromise<int> promise;
+  auto result = promise.Task();
+  const auto work = std::make_shared<RunWork>([&started, &release, promise = std::move(promise)] mutable {
+    started.Notify();
+    // A queued wrapper may finish while its caller-owned read is still running.
+    // The coordinator owns that inline lifetime, not the wrapper's completion.
+    promise.SetValue(release.WaitForNotificationWithTimeout(absl::Seconds(5)) ? 42 : 0);
+  });
+  auto wrapper = executor.Submit([&started, &release, work] {
+    const bool caller_started = started.WaitForNotificationWithTimeout(absl::Seconds(5));
+    work->Run();
+    release.Notify();
+    return caller_started;
+  });
+  work->Run();
+  EXPECT_THAT(wrapper.Get(), IsTrue());
+  EXPECT_THAT(result.Get(), Eq(42));
+}
+
 TEST_F(RunExecutorTest, ReadAheadSlotsAndBytesAreSharedAndReleasedByTheirOwner) {
   RunExecutor executor(2, 100);
   EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt));