Update: simplify NEXT_LEVEL scheduling with explicit targets

docs/callable-identity-registration.md

@@ -342,11 +342,11 @@ every active child endpoint in the handle's `target_namespace` for this
 `Worker` has installed the callable identity. Registering to a user-selected
 worker subset is not part of this contract.
 `orch.submit_next_level(..., worker=...)` and
-`orch.submit_next_level_group(..., workers=...)` are submit-time affinity
-controls; they do not define registration scope. NEXT_LEVEL affinity consumes
-stable worker ids. Local Python Worker children and remote L3 workers use the
-worker ids returned by `add_worker(...)` / `add_remote_worker(...)`; L3 chip
-worker ids are the existing chip worker ids.
+`orch.submit_next_level_group(..., workers=...)` require exact submit-time
+placement; they do not define registration scope. NEXT_LEVEL placement uses
+stable worker ids. Local Python Worker children and remote L3 workers use ids
+returned by `add_worker(...)` / `add_remote_worker(...)`; L3 chip worker ids
+are the existing chip worker ids.
 
 `CallableHandle` is the public callable token returned by registration:
 
@@ -381,7 +381,7 @@ matmul = worker.register(chip_callable)
 postprocess = worker.register(py_callable)
 
 def parent_orch(orch, args, config):
-    orch.submit_next_level(matmul, args, config)
+    orch.submit_next_level(matmul, args, config, worker=0)
     orch.submit_sub(postprocess, args)
 ```
 
@@ -488,11 +488,10 @@ Worker teardown is deferred to a later design.
 
 ### Dispatch Contract
 
-Parent-side scheduling assumes the handle's `hashid` is installed on every
-active target in its registration scope. Dispatch choices are constrained by
-the handle namespace, submit-time affinity, and tensor/buffer accessibility.
-Submit-time live validation is a preflight check only. It does not pin the
-target identity through later drain or child dispatch. Callers must not
+Parent-side scheduling requires the handle's `hashid` on the submitted exact
+target. The Orchestrator validates the handle namespace and tensor/buffer
+accessibility for that target before committing the slot. The target identity
+is then fixed in `TaskSlotState` through dispatch. Callers must not
 concurrently unregister a handle while `Worker.run()` or any in-flight task may
 submit or use that handle; wait for the relevant run/drain to return before
 unregistering it.

docs/directed-next-level-scheduling.md

@@ -0,0 +1,79 @@
+# Directed NEXT_LEVEL Scheduling
+
+Every NEXT_LEVEL task names its exact worker at submission time, and the
+Scheduler dispatches it only to that worker. The runtime never selects a
+NEXT_LEVEL worker on the caller's behalf. SUB tasks keep free scheduling and
+take no worker id.
+
+This page is the placement contract. For the queue topology, dispatch loop,
+and dependency/failure handling see [scheduler.md](scheduler.md); for the
+submit API see [orchestrator.md](orchestrator.md).
+
+## Placement contract
+
+- `Orchestrator.submit_next_level(callable, args, config, worker=id)` targets
+  exactly one stable NEXT_LEVEL worker id.
+- `Orchestrator.submit_next_level_group(callable, args_list, config,
+  workers=[...])` takes one stable worker id per member. The ids must be
+  distinct and their count must equal the member count.
+- A worker id must be non-negative and name a registered NEXT_LEVEL worker.
+  Booleans, floats, numeric strings, and other coercible non-integers are
+  rejected.
+- A local (`LOCAL_PYTHON` / `LOCAL_CHIP`) callable must target a local child;
+  a remote worker id is rejected because a remote endpoint installs only its
+  own dispatcher callables.
+- When a callable or its tensor data constrains eligibility (remote callables,
+  remote-buffer data), the target must lie in that eligible set. Eligibility
+  bounds which targets are *valid*; it is not a scheduler selection policy.
+- `submit_sub` / `submit_sub_group` are unchanged and take no worker id.
+
+## Queue and dispatch model
+
+The worker owns a directed-queue registry fixed at `init()` from the stable
+NEXT_LEVEL worker ids:
+
+| Task | Queue |
+| ---- | ----- |
+| NEXT_LEVEL single | `FIFO[target_worker_id]` |
+| NEXT_LEVEL group | one shared group FIFO |
+| SUB | one shared SUB FIFO |
+
+A task's target and queue never change after submission. Tasks that are READY
+at submission and tasks released by a completing dependency use the same
+routing operation.
+
+Each Scheduler iteration:
+
+- launches the group FIFO head only when every one of its target workers is
+  idle — all members or none, with no partial reservation;
+- dispatches the head of each idle worker's single FIFO;
+- dispatches READY SUB work through the existing free-selection path.
+
+Only the group FIFO head is examined, and a launchable group is tried before
+conflicting singles. The runtime adds no fairness, aging, priority, or
+reservation policy; callers choose worker sets that make acceptable progress.
+
+## Invariants
+
+- PENDING tasks live in task-slot state, not in a ready queue; a task is routed
+  only once all producers complete successfully.
+- A failed producer poisons its dependents instead of enqueuing them.
+- A group is one DAG node and completes only after every member reaches a
+  terminal state.
+- The NEXT_LEVEL worker set is fixed after `init()`; a worker lookup that fails
+  after submit-time validation is an invariant violation, not a fallback.
+- No queue or scheduler mutex is held while an endpoint executes.
+
+## Non-goals
+
+- No SUB worker selection.
+- No priorities, work stealing, rebinding, queue scanning, aging, quotas,
+  preemption, starvation prevention, or partial group reservation.
+- No compatibility flags, environment variables, or macros.
+
+## Related documents
+
+- [scheduler.md](scheduler.md) — queue topology, dispatch loop, completion
+- [orchestrator.md](orchestrator.md) — submit API and DAG construction
+- [remote-l3-worker-design.md](remote-l3-worker-design.md) — remote NEXT_LEVEL
+  workers

docs/hierarchical_level_runtime.md

@@ -86,23 +86,24 @@ Owns:
   identity and logical offset.
 - `Scope` — lifetime management for intermediate tensors
 
-One `submit_next_level(callable, task_args, config)` call:
+One `submit_next_level(callable, task_args, config, worker=worker_id)` call:
 
 1. allocates a slot
 2. moves task data into the slot
 3. walks `TaskArgs` tags (INPUT/OUTPUT/INOUT/OUTPUT_EXISTING/NO_DEP) to
    lookup/insert TensorMap entries
 4. records fanin metadata on producer slots
-5. pushes the new slot onto the scheduler's wiring queue
+5. attaches fanout edges and routes READY work to its target queue
 
 See [orchestrator.md](orchestrator.md) for the 7-step submit flow and state machine.
 
 ### Scheduler (Scheduler thread)
 
-The **DAG executor**. A dedicated C++ thread that drains three queues:
+The **DAG executor**. A dedicated C++ thread that handles:
 
-- **wiring queue** — slots just submitted; wire fanout edges, compute readiness
-- **ready queue** — slots with all fanin satisfied; pick an idle WorkerThread and dispatch
+- **directed NEXT_LEVEL queues** — one single-task FIFO per stable worker ID
+  plus one all-targets-ready group FIFO
+- **shared SUB queue** — freely select idle SUB WorkerThreads
 - **completion queue** — slots whose worker finished; release fanout, wake downstream consumers, retire slot
 
 The Scheduler never inspects task data — it just moves slot ids between queues
@@ -141,13 +142,13 @@ what flows through `ChipWorker::run`.
                  │ submit(callable, args, config)  │
                  │   1. ring.alloc()               │
                  │   2. TensorMap lookup/insert    │
-                 │   3. record fanin              │
-                 │   4. push wiring_queue ───────►│
-                 │                                 │ Phase 0: drain wiring_queue
-                 │                                 │   wire fanout edges
-                 │                                 │   if ready → ready_queue
-                 │                                 │ pop ready_queue
-                 │                                 │ pick idle WorkerThread
+                 │   3. attach fanout + fanin      │
+                 │   4. if READY, enqueue directed/shared ready queue
+                 │   5. notify Scheduler ─────────►│ wake Scheduler
+                 │                                 │ drain completion_queue
+                 │                                 │   release/poison fanout
+                 │                                 │ pop directed/shared queue
+                 │                                 │ resolve target (NL) or idle SUB
                  │                                 │ wt.dispatch(slot_id) ──────► WorkerThread
                  │                                 │                              encode mailbox → spin-poll TASK_DONE
                  │                                 │                              (blocking; child runs the kernel)
@@ -163,7 +164,8 @@ Communication channels:
 
 | Path | Mechanism | Payload |
 | ---- | --------- | ------- |
-| Orch → Scheduler | wiring_queue (mutex + CV) | slot id |
+| Orchestrator/Scheduler → ready queues | direct `enqueue_ready` queue push | slot id |
+| Orchestrator/Scheduler → Scheduler loop | condition-variable notification | ready wake-up |
 | Scheduler → WorkerThread | WorkerThread internal queue | slot id |
 | WorkerThread → Scheduler | completion_queue (mutex + CV) | slot id + group index + outcome |
 | WorkerThread ↔ child | shm mailbox (state + error + task data) | encoded blob |

docs/orchestrator.md

@@ -34,13 +34,13 @@ public:
     SubmitResult submit_next_level(const CallableIdentity &callable,
                                     const TaskArgs &args,
                                     const CallConfig &config,
-                                    int32_t worker = -1,
+                                    int32_t worker,
                                     const std::vector<int32_t> &eligible_worker_ids = {},
                                     const RemoteTaskArgsSidecar &remote_sidecar = {});
     SubmitResult submit_next_level_group(const CallableIdentity &callable,
                                           const std::vector<TaskArgs> &args_list,
                                           const CallConfig &config,
-                                          const std::vector<int32_t> &workers = {},
+                                          const std::vector<int32_t> &workers,
                                           const std::vector<std::vector<int32_t>> &eligible_worker_ids = {},
                                           const std::vector<RemoteTaskArgsSidecar> &remote_sidecars = {});
     SubmitResult submit_sub(const CallableIdentity &callable,
@@ -75,11 +75,12 @@ Remote L3 submit adds two hidden pieces of metadata: final eligible worker-id
 sets and optional `RemoteTaskArgsSidecar` entries aligned by tensor index.
 Python `RemoteCallable` handles supply callable eligibility, and
 `TaskArgs.add_tensor(RemoteTensorRef(...), tag)` supplies tensor sidecars. The
-Orchestrator validates affinity, worker existence, local-vs-remote
+Orchestrator validates exact placement, worker existence, local-vs-remote
 compatibility, remote handle access rights for the tensor tag, bare host
 pointers, and remote null OUTPUT tensors before committing the slot.
-For NEXT_LEVEL tasks, `worker`/`workers` are stable worker ids rather than
-C++ worker-thread vector indices.
+For NEXT_LEVEL tasks, `worker`/`workers` are required stable worker ids rather
+than C++ worker-thread vector indices. SUB submit APIs expose no worker
+selection and use the shared SUB ready queue.
 
 ---
 
@@ -92,7 +93,8 @@ how the slot is set up.
 ```cpp
 SubmitResult Orchestrator::submit_next_level(const CallableIdentity &callable,
                                               TaskArgs args,
-                                              const CallConfig &config) {
+                                              const CallConfig &config,
+                                              int32_t worker) {
     // 1. Alloc slot (blocks on back-pressure if ring full)
     TaskSlot sid = ring_.alloc();
     TaskSlotState &s = slots_[sid];
@@ -103,6 +105,7 @@ SubmitResult Orchestrator::submit_next_level(const CallableIdentity &callable,
     s.callable    = callable;
     s.task_args   = std::move(args);
     s.config      = config;
+    s.target_worker_ids = {worker};
 
     // 3. Walk task_args tags, derive dependencies
     //    (dedup producers: same producer may appear on multiple input tensors)
@@ -130,10 +133,16 @@ SubmitResult Orchestrator::submit_next_level(const CallableIdentity &callable,
     // 5. Register with scope (holds slot open until scope_end releases ref)
     scope_.register_task(sid);          // increments s.fanout_total by 1
 
-    // 6. Push fanout edges onto scheduler's wiring queue
-    //    (Scheduler wires producer→consumer asynchronously; avoids blocking
-    //    the Orch thread on fanout_mu)
-    scheduler_.enqueue_wiring(sid, std::move(producers));
+    // 6. Attach fanout edges under each producer's mutex. Producers already
+    //    completed do not count as live fanins; failed producers poison this
+    //    slot. Route an immediately READY slot through enqueue_ready().
+    attach_fanout_and_count_live_producers(sid, producers);
+    if (s.fanin_count == 0) {
+        s.state = TaskState::READY;
+        enqueue_ready(sid);
+    } else {
+        s.state = TaskState::PENDING;
+    }
 
     // 7. Return handle
     return {sid};
@@ -172,16 +181,21 @@ remote buffer identity and logical offset.
 
 **Step 4 — fanin count**: The number of live producers. Decremented by
 `fanin_released++` each time a producer completes; when `fanin_released ==
-fanin_count`, the slot is ready.
+fanin_count`, the slot is ready. A ready NEXT_LEVEL single task is routed to
+the FIFO for its required stable worker id. The same routing function is used
+for immediately-ready submissions and tasks released by Scheduler dependency
+processing. A ready NEXT_LEVEL group is routed to the dedicated group FIFO;
+SUB tasks remain on their shared queue.
 
 **Step 5 — scope ref**: Each slot starts with one "scope reference" in its
 fanout_total. Without this, a task with no downstream consumer would never be
 reclaimable. See [§6 Scope](#6-scope).
 
-**Step 6 — wiring queue**: Fanout edges (producer knows its consumers) are
-wired **asynchronously** by the Scheduler thread. This decouples submit from
-`fanout_mu` contention. See [scheduler.md](scheduler.md) §2 for the wiring
-phase.
+**Step 6 — fanout attachment and READY routing**: Submission synchronously
+locks each producer's `fanout_mu`, attaches the consumer, and counts only live
+producers. An immediately READY task is routed to its exact NEXT_LEVEL worker
+FIFO, the NEXT_LEVEL group FIFO, or the shared SUB FIFO. See
+[scheduler.md](scheduler.md) §1.
 
 ---
 
@@ -192,49 +206,29 @@ Each worker gets its own `TaskArgs`; the node only reaches COMPLETED when all
 N finish.
 
 ```cpp
-SubmitResult Orchestrator::submit_next_level_group(const CallableIdentity &callable,
-                                                    std::vector<TaskArgs> args_list,
-                                                    const CallConfig &config) {
-    TaskSlot sid = ring_.alloc();
-    TaskSlotState &s = slots_[sid];
-    s.reset();
-    s.worker_type     = WorkerType::NEXT_LEVEL;
-    s.callable        = callable;
-    s.config          = config;
-    s.group_size      = args_list.size();
-    s.sub_complete_count = 0;
-    s.task_args_list  = std::move(args_list);
-
-    // Tag walk unions all entries in args_list (any input in any member → fanin)
-    // Dedup both producers and outputs across all args_list entries.
-    std::vector<TaskSlot> producers;
-    std::unordered_set<TaskSlot> producers_seen;
-    std::unordered_set<uint64_t> outputs_seen;
-    for (auto &a : s.task_args_list) {
-        for (int i = 0; i < a.tensor_count(); i++) {
-            TensorArgType tag = a.tag(i);
-            uint64_t ptr      = a.tensor(i).data;
-            if (tag == INPUT || tag == INOUT)
-                if (auto prod = tensormap_.lookup(ptr); prod != INVALID)
-                    if (producers_seen.insert(prod).second)
-                        producers.push_back(prod);
-            if (tag == OUTPUT || tag == INOUT || tag == OUTPUT_EXISTING)
-                if (outputs_seen.insert(ptr).second)
-                    tensormap_.insert(ptr, sid);
-        }
-    }
-
-    s.fanin_count    = static_cast<int32_t>(producers.size());
-    s.fanin_released = 0;
-    scope_.register_task(sid);
-    scheduler_.enqueue_wiring(sid, std::move(producers));
-    return {sid};
+SubmitResult Orchestrator::submit_next_level_group(
+    const CallableIdentity &callable, const std::vector<TaskArgs> &args_list,
+    const CallConfig &config, const std::vector<int32_t> &worker_ids,
+    const std::vector<std::vector<int32_t>> &eligible_worker_ids,
+    const std::vector<RemoteTaskArgsSidecar> &remote_sidecars
+) {
+    return submit_impl(
+        WorkerType::NEXT_LEVEL, callable, config, args_list, worker_ids,
+        eligible_worker_ids, remote_sidecars
+    );
 }
 ```
 
-At dispatch time the Scheduler reserves `group_size` idle WorkerThreads, and
-each WorkerThread runs `worker->run` with its own `task_args_list[i]`.
-Completion is gated on `sub_complete_count.fetch_add(1) + 1 == group_size`.
+`submit_impl` validates that `worker_ids` contains one unique, eligible target
+per group member before it performs shared dependency inference and READY
+routing.
+
+At dispatch time the Scheduler checks the group FIFO head and resolves every
+entry in `workers` to that exact stable worker ID. It dispatches only if the
+entire target set is idle; a blocked group reserves no partial worker set and
+does not cause a scan past the FIFO head. Each WorkerThread runs `worker->run`
+with its own `task_args_list[i]`. Completion remains aggregated at the group
+slot, so downstream consumers are released once after every member is terminal.
 
 ---
 
@@ -443,11 +437,11 @@ Flow:
 ```python
 def my_orch(orch, args, cfg):
     with orch.scope():                             # ring 1
-        orch.submit_next_level(chip_a, a_args, cfg)
-        orch.submit_next_level(chip_b, b_args, cfg)
+        orch.submit_next_level(chip_a, a_args, cfg, worker=0)
+        orch.submit_next_level(chip_b, b_args, cfg, worker=1)
     # Inner tasks are now eligible for reclamation on ring 1,
     # without waiting for any outer-scope task.
-    orch.submit_next_level(chip_c, c_args, cfg)    # ring 0 (outer)
+    orch.submit_next_level(chip_c, c_args, cfg, worker=0)  # ring 0
 ```
 
 `with orch.scope():` is the recommended form. Raw `orch.scope_begin()` /
@@ -722,8 +716,8 @@ instead of stalling forever. Default timeout: 10 s.
 
 - [hierarchical_level_runtime.md](hierarchical_level_runtime.md) — how
   Orchestrator fits alongside Scheduler and Worker
-- [scheduler.md](scheduler.md) — what happens to slots after they're pushed
-  onto the wiring queue
+- [scheduler.md](scheduler.md) — READY dispatch and completion-time dependency
+  release
 - [task-flow.md](task-flow.md) — the data (Callable / TaskArgs / CallConfig)
   being moved by `submit_*`
 - [comm-domain.md](comm-domain.md) — `orch.allocate_domain` dynamic