The README now leads with what you actually get: a Python asyncio sur…

README.md

@@ -1,8 +1,21 @@
 # rust-python-lab
 
-A Tokio-backed **execution controller** for Python AI/agent apps. Python stays the authoring surface (`await handle`, `async with`, `async for`). Rust owns admission, concurrency, timeouts, cancellation, retries, streaming, and shutdown. Python's asyncio event loop still executes coroutine bodies — LLM SDKs, HTTP clients, DB drivers, and tools.
+Keep writing agents in Python. Let **Rust run the scheduler**.
 
-This is not a LangChain/LangGraph rewrite and not a Rust HTTP/LLM client. Rust waits on bridged asyncio futures; it does not step Python bytecode.
+`rust_python_lab` is a Tokio-backed execution controller for asyncio apps. You `submit`, `gather`, `stream`, and `await` as usual. Rust owns concurrency, backpressure, timeouts, cancellation, retries, and shutdown — without rewriting your HTTP, DB, or LLM calls, and without locking you into LangChain or LangGraph.
+
+Python still steps coroutine bodies. Rust does not execute Python bytecode. That is the point: the control plane stays off the asyncio thread, so a slow or greedy task cannot stall admission, timers, or cancel.
+
+## Why this exists
+
+- **Python-native API.** `await handle`, `async with Runtime`, `async for` completions and tokens. `submit` is synchronous so you can `cancel()` before the first `await`.
+- **Real cancel and timeout.** Dropping a Tokio waiter is not enough. The runtime cancels the **asyncio Task** the bridge created, so httpx connections and tool calls actually stop.
+- **Limits that hold under load.** `max_concurrency` is a Tokio semaphore. The admission queue is bounded. `OnFull.REJECT` raises `QueueFull`; `OnFull.WAIT` blocks until a slot exists (and aborts on shutdown).
+- **Retries that work.** Rust re-invokes the Python callable for a **new** coroutine. A consumed coroutine object is never replayed.
+- **Fan-out without a Python control-plane.** `gather`, `as_completed`, and bounded `stream()` keep in-flight state in Rust. `stats()` is O(1) counters, not a walk of Python objects.
+- **Clean shutdown.** `await rt.shutdown(timeout=…)` stops new work, drains, then cancels the rest. `async with` does this on exit. Optional `idle_ttl` closes admission after quiet. Further `submit` raises `RuntimeClosed`.
+- **Use the libraries you already have.** httpx, OpenAI/Anthropic SDKs, databases, tools — they stay asyncio. Several `Runtime` instances can apply different policies (for example an LLM pool of 8 and a DB pool of 64).
+- **Proven on sockets, not `asyncio.sleep`.** CI drives a local HTTP/SSE server through httpx: concurrency caps, handler abort on timeout/cancel, retries, backpressure, streams, shutdown, and a short soak.
 
 ## Execution split
 
@@ -12,11 +25,9 @@ This is not a LangChain/LangGraph rewrite and not a Rust HTTP/LLM client. Rust w
 | **Python asyncio** | coroutine bodies, sockets, SDK calls, object lifetimes, GIL |
 | **Bridge** | `pyo3-async-runtimes`: capture `TaskLocals`, convert awaitables both ways, never hold the GIL across `.await` |
 
-Cancel and timeout must cancel the **asyncio Task**, not only drop a Tokio waiter. Retries re-invoke the Python callable to get a new coroutine.
-
 ## Setup
 
-Requires Rust (stable) and Python 3.12. Local development uses 3.12.3.
+Rust (stable) and Python 3.12+.
 
 ```bash
 python -m venv .venv
@@ -40,7 +51,9 @@ async with Runtime(
 ) as rt:
     handle = rt.submit(call_llm, "hello", timeout=10.0)
     result = await handle
-    handle.cancel()
+
+    slow = rt.submit(call_llm, "long job")
+    slow.cancel()
 
     results = await rt.gather(
         [Task(call_llm, prompt) for prompt in prompts],
@@ -51,59 +64,62 @@ async with Runtime(
     async for token in rt.stream(llm_tokens, "hello"):
         ...
 
-    print(rt.stats())
-    await rt.shutdown(timeout=5.0)
+    print(rt.stats())  # queued, in_flight, succeeded, failed, cancelled, timed_out, completed
 ```
 
-After close, `submit` raises `RuntimeClosed`. A full queue with `OnFull.REJECT` raises `QueueFull`.
+`Handle.status()` is `queued`, `running`, `succeeded`, `failed`, `cancelled`, or `timed_out`. After close, `submit` raises `RuntimeClosed`. A full queue with `OnFull.REJECT` raises `QueueFull`. Without `async with`, call `await rt.shutdown(timeout=5.0)` yourself.
 
 ## Test
 
-CI and local default: **localhost HTTP only**, no API keys, no public internet.
+CI and the default local run use **localhost HTTP only** — no API keys, no public internet.
 
 ```bash
 pytest
 ```
 
-The suite drives a local `aiohttp` server (`tests/http_harness.py`) through **httpx** so policy is proven against real sockets:
+The suite uses `tests/http_harness.py` (aiohttp on `127.0.0.1`) and **httpx** task bodies:
 
-- concurrency cap, timeout, and cancel against `/work` and `/slow`
-- retries against `/flaky`
-- `gather` / `as_completed` / SSE `stream` against mixed latencies and `/sse`
-- shutdown cancels in-flight `/slow` and rejects later submits
+- `/work` — peak in-flight ≤ `max_concurrency`
+- `/slow` — timeout and `handle.cancel()` abort the server handler
+- `/flaky` — retries re-call the callable (extra server hits)
+- `/sse` — `rt.stream` plus bounded-buffer backpressure
+- mixed latencies in `gather` / `as_completed`
+- shutdown cancels in-flight `/slow`; later `submit` is `RuntimeClosed`
 
-Short soak (2k tiny GETs; `SOAK=1` runs 20k):
+Soak (2k tiny GETs; `SOAK=1` runs 20k):
 
 ```bash
 pytest tests/test_soak.py
 SOAK=1 pytest tests/test_soak.py
 ```
 
-Agent-shaped example (tools + SSE “LLM”) against the same harness:
+Agent-shaped example (parallel tools + SSE “LLM”) against the same harness:
 
 ```bash
 python examples/agent_tools.py
 ```
 
-Control-plane baseline (not a CI gate): `Runtime.gather` vs `asyncio.gather` + `asyncio.Semaphore` on local HTTP.
+Scheduler baseline (not a CI gate): `Runtime.gather` vs `asyncio.gather` + `asyncio.Semaphore` on local HTTP.
 
 ```bash
 python scripts/bench_fanout.py
 ```
 
 ### Optional live LLM
 
-Skipped unless a key is present. CI never sets keys. Cheap smoke: 2–3 short prompts, `as_completed`, one token stream, one cancelled handle.
+Skipped unless a key is set. CI never sets keys. Cheap smoke: two completions via `as_completed`, one token stream, one cancelled handle. Uses **httpx** only (no official SDK, no LangChain).
 
 ```bash
 python examples/live_llm.py
 ```
 
-Uses `OPENAI_API_KEY` (preferred) or `ANTHROPIC_API_KEY`. Optional `OPENAI_MODEL` / `ANTHROPIC_MODEL`. To also run it under pytest: `LIVE_LLM=1 pytest tests/test_examples.py`. Do not log secrets.
+Set `OPENAI_API_KEY` (preferred) or `ANTHROPIC_API_KEY`. Optional `OPENAI_MODEL` / `ANTHROPIC_MODEL`. To run it under pytest: `LIVE_LLM=1 pytest tests/test_examples.py`. Do not log secrets.
+
+## Honest limits
 
-## What this does not prove
+These are deliberate, not missing checkboxes:
 
-- Rust is not faster at token generation. The model and Python SDK still dominate latency. Benchmarks measure **scheduler** cost on local HTTP, not provider SLA.
+- Rust does not make the model faster. Token generation and the Python SDK still dominate latency. `scripts/bench_fanout.py` measures **scheduler** cost on local HTTP, not provider SLA.
 - No crash-durable journal replay, no multi-process supervisor, no sandbox for tool code.
-- Live tests need a key, cost a few tokens, and will be rate-limited. They are a smoke, not a soak.
-- No built-in Rust HTTP or LLM adapters in this lab. Coroutine bodies stay on asyncio on purpose.
+- Live LLM is an opt-in smoke. It needs a key, spends a few tokens, and will be rate-limited.
+- No built-in Rust HTTP or LLM client. Coroutine bodies stay on asyncio so existing Python libraries keep working.

src/bridge.rs

@@ -9,9 +9,31 @@ use pyo3::types::PyDict;
 use pyo3_async_runtimes::TaskLocals;
 use tokio_util::sync::CancellationToken;
 
-static CANCELLED_EXC: OnceLock<Py<PyAny>> = OnceLock::new();
+pub type PyValue = Py<PyAny>;
+pub type PyValueResult = PyResult<PyValue>;
 
-fn cancelled_type(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
+type PyBound<'py> = Bound<'py, PyAny>;
+type PyTaskCell = Option<PyValue>;
+type PyTaskLock = Mutex<PyTaskCell>;
+pub type SharedPyTask = Arc<PyTaskLock>;
+
+type TaskRx = oneshot::Receiver<PyValue>;
+type ResultRx = oneshot::Receiver<PyValueResult>;
+type TaskTx = oneshot::Sender<PyValue>;
+type ResultTx = oneshot::Sender<PyValueResult>;
+
+pub struct BridgedTask {
+    pub task_rx: TaskRx,
+    pub result_rx: ResultRx,
+}
+
+pub fn shared_py_task() -> SharedPyTask {
+    Arc::new(Mutex::new(None))
+}
+
+static CANCELLED_EXC: OnceLock<PyValue> = OnceLock::new();
+
+fn cancelled_type(py: Python<'_>) -> PyResult<PyBound<'_>> {
     if let Some(cls) = CANCELLED_EXC.get() {
         return Ok(cls.bind(py).clone());
     }
@@ -25,8 +47,8 @@ fn cancelled_type(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
 pub fn start_asyncio_task(
     py: Python<'_>,
     locals: &TaskLocals,
-    awaitable: Bound<'_, PyAny>,
-) -> PyResult<(oneshot::Receiver<Py<PyAny>>, oneshot::Receiver<PyResult<Py<PyAny>>>)> {
+    awaitable: PyBound<'_>,
+) -> PyResult<BridgedTask> {
     let (task_tx, task_rx) = oneshot::channel();
     let (result_tx, result_rx) = oneshot::channel();
     let event_loop = locals.event_loop(py);
@@ -45,13 +67,10 @@ pub fn start_asyncio_task(
     kwargs.set_item("context", locals.context(py))?;
     event_loop.call_method("call_soon_threadsafe", (callback,), Some(&kwargs))?;
 
-    Ok((task_rx, result_rx))
+    Ok(BridgedTask { task_rx, result_rx })
 }
 
-pub fn cancel_asyncio_task(
-    event_loop: &Bound<'_, PyAny>,
-    task: &Bound<'_, PyAny>,
-) -> PyResult<()> {
+pub fn cancel_asyncio_task(event_loop: &PyBound<'_>, task: &PyBound<'_>) -> PyResult<()> {
     let cancel = task.getattr("cancel")?;
     event_loop.call_method1("call_soon_threadsafe", (cancel,))?;
     Ok(())
@@ -85,13 +104,13 @@ pub fn parse_timeout_secs(secs: f64) -> PyResult<Duration> {
 
 /// Cancels the asyncio task if the Tokio waiter is dropped mid-flight.
 pub struct CancelOnDrop {
-    event_loop: Py<PyAny>,
-    py_task: Arc<Mutex<Option<Py<PyAny>>>>,
+    event_loop: PyValue,
+    py_task: SharedPyTask,
     armed: bool,
 }
 
 impl CancelOnDrop {
-    pub fn new(event_loop: Py<PyAny>, py_task: Arc<Mutex<Option<Py<PyAny>>>>) -> Self {
+    pub fn new(event_loop: PyValue, py_task: SharedPyTask) -> Self {
         Self {
             event_loop,
             py_task,
@@ -113,18 +132,18 @@ impl Drop for CancelOnDrop {
             return;
         };
         Python::attach(|py| {
-            let _ = cancel_asyncio_task(&self.event_loop.bind(py), &task.bind(py));
+            let _ = cancel_asyncio_task(self.event_loop.bind(py), task.bind(py));
         });
     }
 }
 
 pub async fn wait_with_policy(
     cancel: CancellationToken,
     timeout: Option<Duration>,
-    result_rx: oneshot::Receiver<PyResult<Py<PyAny>>>,
-    event_loop: Py<PyAny>,
-    py_task: Arc<Mutex<Option<Py<PyAny>>>>,
-) -> PyResult<Py<PyAny>> {
+    result_rx: ResultRx,
+    event_loop: PyValue,
+    py_task: SharedPyTask,
+) -> PyValueResult {
     let mut guard = CancelOnDrop::new(event_loop, py_task);
 
     let result = match timeout {
@@ -151,7 +170,7 @@ pub async fn wait_with_policy(
     result
 }
 
-fn map_oneshot(result: Result<PyResult<Py<PyAny>>, oneshot::Canceled>) -> PyResult<Py<PyAny>> {
+fn map_oneshot(result: Result<PyValueResult, oneshot::Canceled>) -> PyValueResult {
     match result {
         Ok(inner) => inner,
         Err(_) => Err(cancelled_error()),
@@ -160,10 +179,10 @@ fn map_oneshot(result: Result<PyResult<Py<PyAny>>, oneshot::Canceled>) -> PyResu
 
 #[pyclass]
 struct CaptureTask {
-    awaitable: Py<PyAny>,
-    event_loop: Py<PyAny>,
-    task_tx: Option<oneshot::Sender<Py<PyAny>>>,
-    result_tx: Option<oneshot::Sender<PyResult<Py<PyAny>>>>,
+    awaitable: PyValue,
+    event_loop: PyValue,
+    task_tx: Option<TaskTx>,
+    result_tx: Option<ResultTx>,
 }
 
 #[pymethods]
@@ -189,12 +208,12 @@ impl CaptureTask {
 
 #[pyclass]
 struct ResultCompleter {
-    tx: Option<oneshot::Sender<PyResult<Py<PyAny>>>>,
+    tx: Option<ResultTx>,
 }
 
 #[pymethods]
 impl ResultCompleter {
-    fn __call__(&mut self, task: Bound<'_, PyAny>) -> PyResult<()> {
+    fn __call__(&mut self, task: PyBound<'_>) -> PyResult<()> {
         let result = match task.call_method0("result") {
             Ok(value) => Ok(value.unbind()),
             Err(err) => Err(err),

src/handle.rs

@@ -1,28 +1,28 @@
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 use pyo3::prelude::*;
 use tokio_util::sync::CancellationToken;
 
-use crate::bridge::cancel_asyncio_task;
+use crate::bridge::{PyValue, SharedPyTask, cancel_asyncio_task};
 use crate::journal::TaskSlot;
 
 /// Submitted work. Await this object to get the Python coroutine's result.
 #[pyclass(frozen)]
 pub struct Handle {
-    awaitable: Py<PyAny>,
+    awaitable: PyValue,
     slot: Arc<TaskSlot>,
     cancel: CancellationToken,
-    event_loop: Py<PyAny>,
-    py_task: Arc<Mutex<Option<Py<PyAny>>>>,
+    event_loop: PyValue,
+    py_task: SharedPyTask,
 }
 
 impl Handle {
     pub fn new(
-        awaitable: Py<PyAny>,
+        awaitable: PyValue,
         slot: Arc<TaskSlot>,
         cancel: CancellationToken,
-        event_loop: Py<PyAny>,
-        py_task: Arc<Mutex<Option<Py<PyAny>>>>,
+        event_loop: PyValue,
+        py_task: SharedPyTask,
     ) -> Self {
         Self {
             awaitable,
@@ -33,7 +33,7 @@ impl Handle {
         }
     }
 
-    pub(crate) fn clone_awaitable(&self, py: Python<'_>) -> Py<PyAny> {
+    pub(crate) fn clone_awaitable(&self, py: Python<'_>) -> PyValue {
         self.awaitable.clone_ref(py)
     }
 }
@@ -47,10 +47,13 @@ impl Handle {
     /// Cancel this task: Tokio token and the asyncio Task, if it has started.
     pub(crate) fn cancel(&self, py: Python<'_>) -> PyResult<()> {
         self.cancel.cancel();
-        if let Some(task) = self.py_task.lock().ok().and_then(|guard| {
-            guard.as_ref().map(|task| task.clone_ref(py))
-        }) {
-            cancel_asyncio_task(&self.event_loop.bind(py), &task.bind(py))?;
+        if let Some(task) = self
+            .py_task
+            .lock()
+            .ok()
+            .and_then(|guard| guard.as_ref().map(|task| task.clone_ref(py)))
+        {
+            cancel_asyncio_task(self.event_loop.bind(py), task.bind(py))?;
         }
         Ok(())
     }

src/journal.rs

@@ -1,5 +1,5 @@
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
+use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
 
 #[repr(u8)]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -64,12 +64,7 @@ impl TaskSlot {
 
     fn transition(&self, from: TaskState, to: TaskState) -> bool {
         self.state
-            .compare_exchange(
-                from as u8,
-                to as u8,
-                Ordering::AcqRel,
-                Ordering::Acquire,
-            )
+            .compare_exchange(from as u8, to as u8, Ordering::AcqRel, Ordering::Acquire)
             .is_ok()
     }
 

src/policy.rs

@@ -6,8 +6,10 @@ use pyo3::prelude::*;
 #[pyclass(eq, eq_int, frozen, from_py_object, name = "OnFull")]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum OnFull {
-    REJECT,
-    WAIT,
+    #[pyo3(name = "REJECT")]
+    Reject,
+    #[pyo3(name = "WAIT")]
+    Wait,
 }
 
 #[pyclass(frozen, from_py_object)]