--- src/lib.rs.before
+++ src/lib.rs
@@ -19,13 +19,24 @@
 type Link = Rc<RefCell<Connection>>;
 type Request = Rc<RefCell<RequestState>>;
 
-fn ready(py: Python<'_>, value: Py<PyAny>, state: Option<&State>) -> PyResult<Py<PyAny>> {
-    let yield_first = state.is_some_and(|s| {
-        let n = s.sends.get() + 1;
-        s.sends.set(if n >= 128 { 0 } else { n });
-        n >= 128
-    });
-    completed::value(py, Some(value), yield_first, state)
+// Fairness belongs to the producing request. Unrelated short requests must
+// not spend each other's quota and delay cleanup after an already-written body.
+fn cooperate(request: &Request) -> bool {
+    let mut r = request.borrow_mut();
+    r.operations += 1;
+    if r.operations < 128 {
+        return false;
+    }
+    r.operations = 0;
+    true
+}
+fn ready(
+    py: Python<'_>,
+    value: Py<PyAny>,
+    request: &Request,
+    state: Option<&State>,
+) -> PyResult<Py<PyAny>> {
+    completed::value(py, Some(value), cooperate(request), state)
 }
 fn message(py: Python<'_>, kind: &str, body: Option<(Py<PyBytes>, bool)>) -> PyResult<Py<PyAny>> {
     let d = PyDict::new(py);
@@ -156,6 +167,7 @@
     head: bool,
     minor: u8,
     task_done: bool,
+    operations: usize,
     trailers: bool,
     body_finished: bool,
     trailers_started: bool,
@@ -173,7 +185,7 @@
     connections: RefCell<HashMap<u64, Link>>,
     tasks: RefCell<HashMap<u64, Py<PyAny>>>,
     next: Cell<u64>,
-    sends: Cell<usize>,
+    starts: RefCell<HashMap<u64, Py<Start>>>,
     stopping: Cell<bool>,
     date: RefCell<Vec<u8>>,
     date_at: Cell<u64>,
@@ -276,6 +288,10 @@
                     "set_exception",
                     (PyOSError::new_err("Client disconnected").value(py),),
                 )?;
+            }
+            let start = self.starts.borrow_mut().remove(&id);
+            if let Some(start) = start {
+                start.borrow(py).discard(py)?;
             }
             let task = self.tasks.borrow().get(&id).map(|t| t.clone_ref(py));
             if let Some(t) = task {
@@ -570,6 +586,7 @@
             head: h.method == "HEAD",
             minor: h.minor,
             task_done: false,
+            operations: 0,
             trailers: false,
             body_finished: false,
             trailers_started: false,
@@ -590,7 +607,35 @@
             },
         )?;
         let result = (|| -> PyResult<()> {
-            let coro = self.app.bind(py).call1((scope, receive, send))?;
+            let coro = self.app.bind(py).call1((scope, receive, send))?.unbind();
+            let start = Py::new(
+                py,
+                Start {
+                    state: Rc::downgrade(self),
+                    request: r,
+                    coroutine: RefCell::new(Some(coro)),
+                },
+            )?;
+            self.starts.borrow_mut().insert(id, start.clone_ref(py));
+            self.loop_
+                .bind(py)
+                .call_method1(pyo3::intern!(py, "call_soon"), (start,))?;
+            Ok(())
+        })();
+        if let Err(e) = result {
+            e.print(py);
+            self.error(py, c, 500)?;
+        }
+        Ok(())
+    }
+    fn launch(
+        self: &Rc<Self>,
+        py: Python<'_>,
+        c: &Link,
+        request: &Request,
+        coro: Py<PyAny>,
+    ) -> PyResult<()> {
+        let result = (|| -> PyResult<()> {
             // Task's eager mode runs until the first real wait. Preserve embedding
             // applications' custom task factories by using their normal loop path.
             let task = if self
@@ -609,19 +654,21 @@
             };
             let done = Done {
                 state: Rc::downgrade(self),
-                request: r.clone(),
+                request: request.clone(),
             };
             if task.call_method0(pyo3::intern!(py, "done"))?.is_truthy()? {
                 // An eager task can be finished without allocating another Python
-                // callback or scheduling another event-loop turn. The outer pump
-                // owns pipeline progress, so this path must not recurse into it.
-                done.finish(py, &task, false)?;
+                // callback or scheduling another event-loop turn. Pipeline progress
+                // queues the next Start; it cannot recurse into application code.
+                done.finish(py, &task, true)?;
             } else {
                 task.call_method1(
                     pyo3::intern!(py, "add_done_callback"),
                     (Py::new(py, done)?,),
                 )?;
-                self.tasks.borrow_mut().insert(id, task.unbind());
+                self.tasks
+                    .borrow_mut()
+                    .insert(request.borrow().id, task.unbind());
             }
             Ok(())
         })();
@@ -1000,10 +1047,7 @@
         } else {
             return Err(PyValueError::new_err("Unsupported ASGI message"));
         }
-        let n = s.sends.get() + 1;
-        s.sends.set(n);
-        if n >= 128 {
-            s.sends.set(0);
+        if cooperate(&self.request) {
             return completed::value(py, None, true, Some(&s));
         }
         Ok(s.done.clone_ref(py))
@@ -1019,13 +1063,28 @@
 impl Receiver {
     fn __call__(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
         let Some(s) = self.state.upgrade() else {
-            return ready(py, message(py, "http.disconnect", None)?, None);
+            return ready(
+                py,
+                message(py, "http.disconnect", None)?,
+                &self.request,
+                None,
+            );
         };
         let Some(c) = self.request.borrow().connection.upgrade() else {
-            return ready(py, message(py, "http.disconnect", None)?, Some(&s));
+            return ready(
+                py,
+                message(py, "http.disconnect", None)?,
+                &self.request,
+                Some(&s),
+            );
         };
         if c.borrow().dead || self.request.borrow().complete || self.request.borrow().task_done {
-            return ready(py, message(py, "http.disconnect", None)?, Some(&s));
+            return ready(
+                py,
+                message(py, "http.disconnect", None)?,
+                &self.request,
+                Some(&s),
+            );
         }
         s.pump(py, &c)?;
         if !self.request.borrow().body.is_empty()
@@ -1033,7 +1092,7 @@
         {
             let msg = take_message(py, &self.request)?;
             s.pump(py, &c)?;
-            return ready(py, msg, Some(&s));
+            return ready(py, msg, &self.request, Some(&s));
         }
         let old = self.request.borrow_mut().receive.take();
         if let Some(f) = old
@@ -1061,6 +1120,44 @@
             .unbind();
         self.request.borrow_mut().receive = Some(f.clone_ref(py));
         Ok(f)
+    }
+}
+
+#[pyclass(unsendable, frozen)]
+struct Start {
+    state: Weak<State>,
+    request: Request,
+    coroutine: RefCell<Option<Py<PyAny>>>,
+}
+impl Start {
+    fn discard(&self, py: Python<'_>) -> PyResult<()> {
+        let coro = self.coroutine.borrow_mut().take();
+        if let Some(coro) = coro {
+            coro.bind(py).call_method0(pyo3::intern!(py, "close"))?;
+        }
+        Ok(())
+    }
+}
+#[pymethods]
+impl Start {
+    fn __call__(&self, py: Python<'_>) -> PyResult<()> {
+        let Some(s) = self.state.upgrade() else {
+            return self.discard(py);
+        };
+        s.starts.borrow_mut().remove(&self.request.borrow().id);
+        let Some(coro) = self.coroutine.borrow_mut().take() else {
+            return Ok(());
+        };
+        let c = self.request.borrow().connection.upgrade();
+        let alive = c.as_ref().is_some_and(|c| !c.borrow().dead);
+        if let Some(c) = c
+            && alive
+        {
+            s.launch(py, &c, &self.request, coro)
+        } else {
+            coro.bind(py).call_method0(pyo3::intern!(py, "close"))?;
+            Ok(())
+        }
     }
 }
 
@@ -1218,7 +1315,7 @@
                 connections: RefCell::new(HashMap::new()),
                 tasks: RefCell::new(HashMap::new()),
                 next: Cell::new(0),
-                sends: Cell::new(0),
+                starts: RefCell::new(HashMap::new()),
                 stopping: Cell::new(false),
                 date: RefCell::new(Vec::new()),
                 date_at: Cell::new(0),
@@ -1236,7 +1333,6 @@
     }
     fn poll(&self, py: Python<'_>) -> PyResult<()> {
         let s = &self.state;
-        s.sends.set(0);
         for (token, mask) in s.poller.events()? {
             if token == 0 {
                 if s.stopping.get() {
@@ -1338,7 +1434,8 @@
         Ok(())
     }
     fn pending(&self) -> usize {
-        self.state.tasks.borrow().len()
+        self.state.starts.borrow().len()
+            + self.state.tasks.borrow().len()
             + self
                 .state
                 .connections