Update: queue single NEXT_LEVEL tasks per worker

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/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. NEXT_LEVEL groups and SUB tasks remain on their shared queues in
+this transition.
 
 **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
@@ -443,11 +448,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()` /

docs/remote-l3-worker-design.md

@@ -137,7 +137,7 @@ Relevant code paths:
     `MAILBOX_OFF_ERROR_MSG`.
 - `src/common/hierarchical/orchestrator.{h,cpp}`
   - `submit_next_level()` stores `TaskArgs`, `CallConfig`,
-  `CallableIdentity`, and optional worker affinity in a parent-side slot.
+  `CallableIdentity`, and the required target worker in a parent-side slot.
   - Dependency inference happens before dispatch from tags in `TaskArgs`.
 - `src/common/task_interface/task_args.h`
   - Process dispatch writes `[T][S][Tensor x T][uint64 x S]`.
@@ -234,9 +234,8 @@ initialization has completed.
 ## Worker Identity and Callable Routing
 
 Remote scheduling needs explicit callable resolver scopes and an explicit
-mapping from callable identities to eligible NEXT_LEVEL workers. The current
-scheduler can otherwise choose any idle worker, which is only correct when
-every NEXT_LEVEL child has the same callable registry.
+mapping from callable identities to eligible NEXT_LEVEL workers. Every submit
+also names the exact worker that will execute the task.
 
 Required contracts:
 
@@ -325,13 +324,10 @@ Required contracts:
 - `TaskSlotState` stores the final eligible worker-id set for the slot. This is
   the intersection of workers that can resolve the callable hashid and workers
   that can access every tensor/buffer referenced by the slot.
-- If the user passes `worker=worker_id`, submit-time validation checks that
-  the worker id is eligible for that hashid and for the slot's tensor
-  sidecars.
-- If `worker=-1`, the Scheduler chooses only from idle workers in the
-  slot's eligible set.
-- Group submit validates each affinity independently. Unconstrained group
-  members are assigned distinct idle eligible endpoints.
+- `worker=worker_id` is required. Submit-time validation checks that the worker
+  id is eligible for that hashid and for the slot's tensor sidecars.
+- Group submit requires one distinct worker id per member and validates each
+  target independently.
 - Mixed local + remote NEXT_LEVEL pools are allowed only when the callable
   hashid is registered on every endpoint that can receive the slot and the
   slot's tensors are materialized in a representation those endpoints can