Update: dispatch NEXT_LEVEL groups exactly

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)
 ```
 

docs/directed-next-level-scheduling/pr1-explicit-placement-contract.md

@@ -0,0 +1,93 @@
+# PR1 Plan: Require Explicit NEXT_LEVEL Placement
+
+## Objective
+
+Make the public NEXT_LEVEL submit contract require a concrete stable worker ID
+for every task. Keep the existing affinity-aware dispatch implementation in
+place so this PR changes the contract and validation without changing queue
+ownership or dependency scheduling.
+
+This PR is the compatibility boundary for the stacked series. After it lands,
+new public submissions cannot request automatic worker selection.
+
+## Scope
+
+- Require `worker=` on `Orchestrator.submit_next_level`.
+- Require `workers=` on `Orchestrator.submit_next_level_group`.
+- Require one worker ID per group member.
+- Reject negative IDs, duplicate group IDs, unknown workers, and targets that
+  are outside the final callable/data eligibility set.
+- Preserve stable NEXT_LEVEL worker-ID semantics for local and remote workers.
+- Update in-repository callers to pass their intended worker IDs.
+- Add focused API and validation tests.
+- Update contract documentation touched by the API change.
+
+## Explicit Non-Goals
+
+- Do not change `submit_sub` or `submit_sub_group`.
+- Do not add SUB worker IDs.
+- Do not change ready-queue topology.
+- Do not remove the scheduler's idle-worker fallback yet.
+- Do not change dependency inference, group completion, failure poisoning,
+  buffer lifetime, or callable registration.
+- Do not add compatibility flags or environment-variable gates.
+
+## Planned File Changes
+
+### Public Python and binding API
+
+- `python/simpler/orchestrator.py`
+  - Make `worker` and `workers` required keyword-only arguments.
+  - Reject negative and duplicate targets before calling C++.
+  - Describe the values as exact targets rather than affinities.
+- `python/bindings/worker_bind.h`
+  - Remove nanobind defaults for `worker` and `workers`.
+  - Update binding help text.
+
+### C++ submit validation
+
+- `src/common/hierarchical/orchestrator.cpp`
+  - Reject missing or negative NEXT_LEVEL targets.
+  - Require group target count to equal group size.
+  - Reject duplicate group targets.
+  - Continue validating target membership in `eligible_worker_ids`.
+- `src/common/hierarchical/orchestrator.h`
+  - Remove public default values for NEXT_LEVEL target arguments.
+  - Update comments to state the exact-placement contract.
+
+### Call sites and tests
+
+- Update examples, scene tests, Python unit tests, and C++ unit tests that rely
+  on the old unconstrained default.
+- Use worker `0` only where the fixture creates exactly one NEXT_LEVEL worker.
+- Use explicit member IDs for group fixtures.
+- Preserve tests for callable/data eligibility by selecting an ID from the
+  final eligible set.
+
+### Documentation
+
+- Update `docs/orchestrator.md`, `docs/scheduler.md`, `docs/task-flow.md`,
+  `docs/hierarchical_level_runtime.md`, and remote-L3 placement text.
+- Remove statements that `-1`, `None`, or an empty list means unconstrained.
+
+## Validation
+
+- Python tests prove missing `worker` and `workers` fail at the API boundary.
+- C++ tests prove negative, unknown, duplicate, count-mismatched, and
+  ineligible targets fail before slot allocation.
+- Existing local and remote explicit-placement tests continue to pass.
+- Relevant examples and scene tests collect with the required arguments.
+
+## Size Budget
+
+- Production code: at most 250 added lines.
+- Tests and call-site updates: at most 350 added lines.
+- Documentation: at most 250 added lines.
+- Total target: at most 850 added lines.
+
+## Completion Criteria
+
+- Every public NEXT_LEVEL submit carries exact target worker IDs.
+- No SUB API or behavior changes.
+- No queue or dispatch algorithm changes.
+- The PR builds and its focused Python/C++ tests pass independently.

docs/directed-next-level-scheduling/pr2-per-worker-single-queues.md

@@ -0,0 +1,93 @@
+# PR2 Plan: Route Single Tasks to Per-Worker Queues
+
+## Objective
+
+Replace the shared NEXT_LEVEL queue for non-group tasks with one FIFO queue per
+stable NEXT_LEVEL worker ID. A single task becomes dispatchable only from the
+queue owned by its submitted target worker.
+
+PR1 is a prerequisite: every NEXT_LEVEL slot must already carry a validated
+target worker ID.
+
+## Scope
+
+- Add a fixed, initialization-time registry of NEXT_LEVEL single-task queues.
+- Route immediately-ready single tasks to their target worker queue.
+- Route dependency-released single tasks to the same target worker queue.
+- Dispatch each queue only to its owning worker.
+- Leave NEXT_LEVEL group tasks on the existing shared NEXT_LEVEL queue.
+- Leave the SUB shared queue and SUB dispatch behavior unchanged.
+
+## Explicit Non-Goals
+
+- Do not add or change public APIs.
+- Do not implement the group ready queue yet.
+- Do not add priorities, work stealing, rebinding, queue scanning, aging, or
+  resource reservation.
+- Do not change the DAG state machine or failure propagation.
+- Do not remove group fallback helpers still needed before PR3.
+
+## Planned Design
+
+Introduce a small queue owner with a fixed mapping:
+
+```text
+stable NEXT_LEVEL worker ID -> ReadyQueue
+```
+
+The mapping is initialized after NEXT_LEVEL workers are registered and before
+the scheduler starts. It is immutable while submissions are allowed, so queue
+lookups need no structural synchronization. Each contained `ReadyQueue`
+remains internally synchronized.
+
+A shared routing helper receives a READY slot and applies this rule:
+
+```text
+NEXT_LEVEL single -> queue[target_worker_id]
+NEXT_LEVEL group  -> legacy NEXT_LEVEL group-capable queue
+SUB               -> existing SUB queue
+```
+
+Both submit-time readiness and completion-time dependency release must call
+the same helper. Duplicating this routing logic is not allowed.
+
+## Planned File Changes
+
+- `src/common/hierarchical/types.h` and `types.cpp`
+  - Add the fixed per-worker queue owner or equivalent focused abstraction.
+- `src/common/hierarchical/worker_manager.h` and `worker_manager.cpp`
+  - Expose the registered NEXT_LEVEL worker IDs needed for initialization.
+- `src/common/hierarchical/worker.h` and `worker.cpp`
+  - Own and initialize the per-worker queues.
+  - Pass them to Orchestrator and Scheduler.
+- `src/common/hierarchical/orchestrator.h` and `orchestrator.cpp`
+  - Route newly READY single slots through the shared routing abstraction.
+- `src/common/hierarchical/scheduler.h` and `scheduler.cpp`
+  - Route newly released single slots identically.
+  - Drain each per-worker queue only when its worker is idle.
+- `tests/ut/cpp/hierarchical/test_orchestrator.cpp`
+  - Verify submit-time routing by exact worker ID.
+- `tests/ut/cpp/hierarchical/test_scheduler.cpp`
+  - Verify dependency-release routing and independent worker progress.
+
+## Validation
+
+- A single task targeted to worker A never dispatches on worker B.
+- A busy worker A does not block a READY single task for idle worker B.
+- A PENDING task enters the correct queue when its final producer completes.
+- A failed producer does not enqueue its consumer.
+- Existing group and SUB tests retain their pre-PR2 behavior.
+
+## Size Budget
+
+- Production code: at most 450 added lines.
+- Tests: at most 350 added lines.
+- Documentation adjustments: at most 100 added lines.
+- Total target: at most 900 added lines.
+
+## Completion Criteria
+
+- All NEXT_LEVEL single dispatch is exact-worker and per-worker FIFO.
+- Group and SUB behavior remains unchanged.
+- Submit-time and completion-time routing share one implementation.
+- The PR builds and focused hierarchical C++ tests pass independently.

docs/directed-next-level-scheduling/pr3-group-ready-queue.md

@@ -0,0 +1,91 @@
+# PR3 Plan: Add Directed Group Dispatch
+
+## Objective
+
+Move NEXT_LEVEL group tasks to a dedicated FIFO queue and dispatch a group only
+when all user-selected workers are idle. Claim and dispatch the complete worker
+set as one scheduler action.
+
+PR1 and PR2 are prerequisites: group members carry validated exact targets,
+and single tasks already use per-worker queues.
+
+## Scope
+
+- Give NEXT_LEVEL groups a dedicated FIFO ready queue.
+- Route immediately-ready and dependency-released groups to that queue.
+- Inspect only the queue head.
+- Resolve every member to its submitted stable worker ID.
+- Dispatch only when all target workers are idle.
+- Dispatch a launchable group before per-worker single queues in the same
+  scheduler iteration.
+- If the head group cannot launch, leave it READY and continue with single
+  queues without reserving any worker.
+- Preserve current group completion aggregation and failure poisoning.
+
+## Explicit Non-Goals
+
+- Do not scan past the group queue head.
+- Do not reserve a subset of group workers.
+- Do not add fairness, aging, priorities, quotas, preemption, or starvation
+  prevention.
+- Do not change SUB group scheduling.
+- Do not change group dependency semantics: one group remains one DAG node.
+
+## Planned Dispatch Rule
+
+```text
+group = group_ready_queue.front
+if group exists and every target worker is idle:
+    pop group
+    mark slot RUNNING
+    dispatch every member to its exact target
+
+for each NEXT_LEVEL worker:
+    if worker is idle:
+        dispatch the head of that worker's single queue
+```
+
+`WorkerThread::dispatch` marks a worker non-idle synchronously. The scheduler
+is the only dispatch producer, so resolving all workers before the first
+dispatch prevents partial launch caused by an ordinary busy-worker check.
+
+## Planned File Changes
+
+- `src/common/hierarchical/types.h` and `types.cpp`
+  - Add the dedicated group queue to the directed queue owner.
+- `src/common/hierarchical/orchestrator.cpp`
+  - Route READY NEXT_LEVEL groups to the group queue.
+- `src/common/hierarchical/scheduler.cpp`
+  - Add launchable-head group dispatch before single dispatch.
+  - Remove idle-pool filling for NEXT_LEVEL groups.
+  - Keep SUB dispatch on its existing path.
+- `tests/ut/cpp/hierarchical/test_scheduler.cpp`
+  - Verify exact member-to-worker mapping.
+  - Verify no member dispatches while one target is busy.
+  - Verify a launchable group wins over conflicting queued singles.
+  - Verify a blocked group does not reserve idle workers.
+  - Verify group completion releases downstream consumers only once.
+
+## Failure and Concurrency Checks
+
+- Duplicate target IDs are rejected by PR1, before queue insertion.
+- Worker lookup failure is treated as an invariant violation, not as a request
+  for fallback selection.
+- Existing group member failure handling remains responsible for terminal
+  aggregation and downstream poison.
+- No new mutex may be held while an endpoint executes.
+
+## Size Budget
+
+- Production code: at most 350 added lines.
+- Tests: at most 400 added lines.
+- Documentation adjustments: at most 100 added lines.
+- Total target: at most 850 added lines.
+
+## Completion Criteria
+
+- NEXT_LEVEL groups use only their exact submitted workers.
+- Group allocation is all-or-nothing at dispatch time.
+- The only cross-queue policy is launchable-group-first.
+- No reservation, scan, rebinding, or fairness mechanism is introduced.
+- Focused group, dependency, and failure tests pass independently.

docs/hierarchical_level_runtime.md

@@ -86,7 +86,7 @@ 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

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,11 @@ 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 remain unconstrained.
 
 ---
 
@@ -92,7 +92,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];
@@ -172,7 +173,11 @@ 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
@@ -232,9 +237,12 @@ SubmitResult Orchestrator::submit_next_level_group(const CallableIdentity &calla
 }
 ```
 
-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`.
+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 +451,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()` /