Add: Worker.submit with per-run completion handles

docs/orchestrator.md

@@ -51,11 +51,12 @@ public:
     // --- Intermediate-buffer allocation (runtime-owned lifetime) ---
     Tensor alloc(const std::vector<uint32_t> &shape, DataType dtype);
 
-    // --- Internal lifecycle (invoked by Worker::run only) ---
+    // --- Internal lifecycle (invoked by Python Worker.submit/RunHandle) ---
     RunId begin_run();
     void close_run_submission(RunId run_id);
     void fail_run_submission(RunId run_id, std::exception_ptr error);
     void wait_run(RunId run_id);
+    bool wait_run_for(RunId run_id, double timeout_seconds);
     bool run_done(RunId run_id) const;
     void release_run(RunId run_id);
     void scope_begin();
@@ -72,9 +73,32 @@ struct SubmitResult { TaskSlot task_slot; };  // internal only; not bound to Pyt
 `config`, since SUB has no per-call config.
 
 The run lifecycle and outer `scope_begin` / `scope_end` are invoked from Python
-`Worker.run` through private bindings. They are not part of the user-facing
-orch-fn API. `Worker.run` remains blocking: it closes submission, waits for the
-matching run fence, performs run-owned cleanup, and returns `None`.
+`Worker.submit` through private bindings. They are not part of the user-facing
+orch-fn API. `Worker.submit` invokes the orchestration callback and closes DAG
+submission synchronously, then returns a `RunHandle` before L3 device work has
+necessarily completed. Graph-construction errors therefore remain synchronous;
+device or endpoint errors are attached to the handle and raised by `wait()` or
+`result()`.
+
+`RunHandle.done` polls the matching native run fence. `wait(timeout)` supports
+bounded waits without cancelling or corrupting the run, and repeated waits
+replay the same terminal result. The handle keeps its `Worker`, callback
+arguments, configuration, and run-owned cleanup state alive until completion.
+`Worker.close()` rejects new submissions and drains every accepted handle
+before tearing down the worker tree.
+
+`Worker.run` remains source-compatible and blocking:
+
+```python
+worker.run(orchestration, args, config)
+# Equivalent to:
+worker.submit(orchestration, args, config).wait()
+```
+
+The current L2 backend is synchronous, so L2 `submit()` executes the existing
+blocking path and returns an already-completed handle. L3 asynchronous return
+does not yet imply overlapping runs: a later submit waits for the prior run's
+fence and cleanup before building the next DAG.
 
 Remote L3 submit adds two hidden pieces of metadata: final eligible worker-id
 sets and optional `RemoteTaskArgsSidecar` entries aligned by tensor index.

python/bindings/worker_bind.h

@@ -367,6 +367,10 @@ inline void bind_worker(nb::module_ &m) {
             "_wait_run", &Orchestrator::wait_run, nb::arg("run_id"), nb::call_guard<nb::gil_scoped_release>(),
             "Block until one run is terminal and raise only that run's error."
         )
+        .def(
+            "_wait_run_for", &Orchestrator::wait_run_for, nb::arg("run_id"), nb::arg("timeout_seconds"),
+            nb::call_guard<nb::gil_scoped_release>(), "Wait up to timeout_seconds for one run to become terminal."
+        )
         .def("_run_done", &Orchestrator::run_done, nb::arg("run_id"))
         .def("_release_run", &Orchestrator::release_run, nb::arg("run_id"));
 

python/simpler/orchestrator.py

@@ -6,7 +6,7 @@
 # INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
 # See LICENSE in the root of the software repository for the full text of the License.
 # -----------------------------------------------------------------------------------------------------------
-"""Orchestrator — DAG builder exposed to the user's orch function during Worker.run().
+"""Orchestrator — DAG builder passed to a Worker submit/run callback.
 
 A thin Python facade over the C++ ``Orchestrator``. The Worker creates one
 Orchestrator handle at init, retrieves the C++ object via ``Worker.get_orchestrator()``,
@@ -24,10 +24,12 @@ def my_orch(orch, args, cfg):
         sub_args.add_tensor(make_tensor_arg(output_tensor), TensorArgType.INPUT)
         orch.submit_sub(sub_handle, sub_args)
 
-    w.run(my_orch, my_args, my_config)
+    handle = w.submit(my_orch, my_args, my_config)
+    handle.wait()
 
-Scope/drain lifecycle is managed by ``Worker.run()``; users never call those
-directly.
+Scope and submission-close lifecycle is managed by ``Worker.submit()``;
+completion is managed by its ``RunHandle``. ``Worker.run()`` remains the
+blocking ``submit(...).wait()`` compatibility entry point.
 """
 
 from __future__ import annotations
@@ -440,7 +442,7 @@ def create_l3_l2_queue(self, *, worker_id: int, depth: int, input_arena_bytes: i
     # deeper heap ring (``min(depth, MAX_RING_DEPTH-1)``) so their
     # memory reclaims independently of the outer scope. ``scope_end`` is
     # non-blocking — it releases scope refs and returns; call
-    # ``Worker.run``/``drain`` for a synchronous wait.
+    # the ``RunHandle`` returned by ``Worker.submit`` for a synchronous wait.
     #
     # Usage::
     #
@@ -543,7 +545,7 @@ def alloc(self, shape: Sequence[int], dtype: DataType) -> Tensor:
         return tensor
 
     # ------------------------------------------------------------------
-    # Internal (called by Worker.run)
+    # Internal (called by Worker.submit)
     # ------------------------------------------------------------------
 
     def _scope_begin(self) -> None:
@@ -564,6 +566,9 @@ def _fail_run_submission(self, run_id: int, message: str = "") -> None:
     def _wait_run(self, run_id: int) -> None:
         self._o._wait_run(run_id)
 
+    def _wait_run_for(self, run_id: int, timeout_seconds: float) -> bool:
+        return bool(self._o._wait_run_for(run_id, timeout_seconds))
+
     def _run_done(self, run_id: int) -> bool:
         return bool(self._o._run_done(run_id))