Bind the handler-dispatch handle to its dispatch window, add GC traversal, fail fast on observer errors, and reclaim arena slots (#492, #497, #500, #506)

packages/doeff-vm-core/src/arena.rs

@@ -1,24 +1,63 @@
 //! Fiber arena for stable fiber IDs within a run.
 
+use std::sync::{Arc, Mutex};
+
 use crate::continuation::{DetachedFiber, DetachedFiberChain};
 use crate::error::VMError;
 use crate::ids::FiberId;
 use crate::segment::Fiber;
 
+/// Channel through which a dropped `DetachedFiberChain` returns its arena
+/// slot indices for reuse (#497).
+///
+/// This is allocator bookkeeping, NOT fiber ownership (ISSUE-VM-001 G1 /
+/// SPEC-VM-021): no Fiber, chain, or continuation ever flows through it.
+/// The chain's fibers move into Continuation ownership at detach and are
+/// destroyed by the chain's own Drop; only the now-permanently-vacant slot
+/// indices are reported here so the arena can return them to its free list
+/// instead of stranding them until run end. Reports may arrive from
+/// arbitrary Python dealloc points, on any thread — hence the Mutex.
+#[derive(Debug, Default)]
+pub struct SlotReclaimQueue {
+    dropped_slot_indices: Mutex<Vec<usize>>,
+}
+
+impl SlotReclaimQueue {
+    /// Called from `DetachedFiberChain::drop` with the slot indices the
+    /// chain still owned when it was abandoned.
+    pub(crate) fn report_dropped_slots(&self, indices: impl Iterator<Item = usize>) {
+        let mut queue = self.dropped_slot_indices.lock().unwrap();
+        queue.extend(indices);
+    }
+
+    /// Drain all pending reports. Cheap when empty (no allocation).
+    fn take_pending(&self) -> Vec<usize> {
+        let mut queue = self.dropped_slot_indices.lock().unwrap();
+        std::mem::take(&mut *queue)
+    }
+}
+
 pub struct FiberArena {
     fibers: Vec<Option<Fiber>>,
     free_list: Vec<usize>,
+    /// Reclaim reports from detached chains dropped without reattachment
+    /// (#497). Drained back into `free_list` before allocating. Replaced
+    /// wholesale on `clear()` so a chain outliving its run session cannot
+    /// poison the next session's free list.
+    slot_reclaim: Arc<SlotReclaimQueue>,
 }
 
 impl FiberArena {
     pub fn new() -> Self {
         FiberArena {
             fibers: Vec::new(),
             free_list: Vec::new(),
+            slot_reclaim: Arc::new(SlotReclaimQueue::default()),
         }
     }
 
     pub fn alloc(&mut self, fiber: Fiber) -> FiberId {
+        self.reclaim_dropped_chain_slots();
         if let Some(idx) = self.free_list.pop() {
             self.fibers[idx] = Some(fiber);
             FiberId::from_index(idx)
@@ -29,6 +68,31 @@ impl FiberArena {
         }
     }
 
+    /// Return slots abandoned by dropped detached chains to the free list.
+    ///
+    /// A detached chain owns its arena slots (single-location law): they
+    /// stay vacant-reserved while the continuation is live. When the chain
+    /// is dropped without reattachment its Drop impl reports the indices to
+    /// `slot_reclaim`; this reconciliation makes them allocatable again
+    /// instead of stranding until `clear()` at run end (#497).
+    pub fn reclaim_dropped_chain_slots(&mut self) {
+        for idx in self.slot_reclaim.take_pending() {
+            #[cfg(feature = "invariant-checks")]
+            {
+                if !matches!(self.fibers.get(idx), Some(None)) {
+                    panic!(
+                        "arena: dropped-chain slot {idx} is not vacant-reserved \
+                         (single-location law violated by reclaim)"
+                    );
+                }
+                if self.free_list.contains(&idx) {
+                    panic!("arena: dropped-chain slot {idx} is already on the free list");
+                }
+            }
+            self.free_list.push(idx);
+        }
+    }
+
     pub fn free(&mut self, id: FiberId) {
         if let Some(slot) = self.fibers.get_mut(id.index()) {
             if slot.take().is_some() {
@@ -58,6 +122,7 @@ impl FiberArena {
 
         let mut chain = DetachedFiberChain::new(head, last_fiber, detached);
         let _ = chain.set_tail_parent(None);
+        chain.arm_slot_reclaim(Arc::clone(&self.slot_reclaim));
         Ok(chain)
     }
 
@@ -151,6 +216,10 @@ impl FiberArena {
     pub fn clear(&mut self) {
         self.fibers.clear();
         self.free_list.clear();
+        // Detach from chains that outlive this session: their late drops
+        // report into the replaced queue, which nobody drains — a stale
+        // index from a previous session must never reach the new free list.
+        self.slot_reclaim = Arc::new(SlotReclaimQueue::default());
     }
 
     pub fn shrink_to_fit(&mut self) {
@@ -350,4 +419,92 @@ mod tests {
             Some(unrelated)
         );
     }
+
+    #[test]
+    fn test_dropped_chain_slots_are_reclaimed_on_next_alloc() {
+        // #497: a detached chain dropped without reattachment (abort-style
+        // handler, scheduler cancellation) must return its slots to the
+        // free list instead of stranding them until run end.
+        let mut arena = FiberArena::new();
+
+        let boundary = arena.alloc(Fiber::new(None));
+        let body = arena.alloc(Fiber::new(Some(boundary)));
+
+        let chain = arena.detach_chain(body, boundary).unwrap();
+        assert_eq!(arena.len(), 0);
+        assert_eq!(arena.slot_count(), 2);
+
+        drop(chain);
+
+        let reused_a = arena.alloc(Fiber::new(None));
+        let reused_b = arena.alloc(Fiber::new(None));
+        assert!(reused_a.index() < 2, "first alloc must reuse a reclaimed slot");
+        assert!(reused_b.index() < 2, "second alloc must reuse a reclaimed slot");
+        assert_eq!(
+            arena.slot_count(),
+            2,
+            "slot vector must not grow after an abandoned chain is dropped"
+        );
+    }
+
+    #[test]
+    fn test_abort_loop_slot_count_is_bounded() {
+        // #497 regression shape: repeated detach-then-drop cycles (one per
+        // abandoned dispatch) must keep the slot vector bounded.
+        let mut arena = FiberArena::new();
+        for _ in 0..100 {
+            let boundary = arena.alloc(Fiber::new(None));
+            let body = arena.alloc(Fiber::new(Some(boundary)));
+            let chain = arena.detach_chain(body, boundary).unwrap();
+            drop(chain);
+        }
+        assert_eq!(arena.len(), 0);
+        assert_eq!(
+            arena.slot_count(),
+            2,
+            "100 abandoned chains must reuse the same two slots"
+        );
+    }
+
+    #[test]
+    fn test_chain_dropped_after_clear_does_not_poison_next_session() {
+        // A chain can outlive its run session (a Python K held across
+        // run()). Its late drop must not inject stale indices into the
+        // next session's free list.
+        let mut arena = FiberArena::new();
+        let boundary = arena.alloc(Fiber::new(None));
+        let body = arena.alloc(Fiber::new(Some(boundary)));
+        let chain = arena.detach_chain(body, boundary).unwrap();
+
+        arena.clear(); // run session ends while the chain is still owned outside
+
+        let live = arena.alloc(Fiber::new(None));
+        drop(chain); // stale report goes to the orphaned queue
+
+        let next = arena.alloc(Fiber::new(None));
+        assert_ne!(next, live, "stale reclaim must not hand out a live slot");
+        assert!(arena.get(live).is_some(), "live fiber must survive stale drops");
+        assert_eq!(arena.len(), 2);
+    }
+
+    #[test]
+    fn test_attached_chain_does_not_reclaim_slots() {
+        // Consuming a chain via attach_chain must NOT report its slots:
+        // the fibers are live in the arena again.
+        let mut arena = FiberArena::new();
+        let boundary = arena.alloc(Fiber::new(None));
+        let body = arena.alloc(Fiber::new(Some(boundary)));
+        let chain = arena.detach_chain(body, boundary).unwrap();
+
+        arena.attach_chain(chain, None).unwrap();
+        arena.reclaim_dropped_chain_slots();
+
+        assert_eq!(arena.len(), 2);
+        let fresh = arena.alloc(Fiber::new(None));
+        assert_eq!(
+            fresh.index(),
+            2,
+            "no slot may be recycled while its fiber is live"
+        );
+    }
 }

packages/doeff-vm-core/src/continuation.rs

@@ -16,11 +16,16 @@
 //! construction: `Option::take()` returns `Some` first time, `None` after.
 //! The VM does not store continuations; for exception recovery during handler
 //! dispatch, it keeps a `Py<PyK>` reference (a Python handle, not a
-//! continuation) — see vm.rs `pending_handler_k_handle`.
+//! continuation) owned by the dispatch itself — a local in
+//! eval_perform/eval_perform_with_k, then `Frame::Program.handler_k_handle`
+//! or `EvalReturnContinuation::ExpandReturn.handler_k_handle` (#492).
+
+use std::sync::Arc;
 
 use pyo3::prelude::*;
 use pyo3::types::PyDict;
 
+use crate::arena::SlotReclaimQueue;
 use crate::ids::FiberId;
 use crate::ir_stream::StreamSourceLocation;
 use crate::memory_stats;
@@ -55,6 +60,14 @@ pub struct DetachedFiberChain {
     head: FiberId,
     last_fiber: FiberId,
     fibers: Vec<DetachedFiber>,
+    /// Where to report still-owned arena slot indices if this chain is
+    /// dropped without being reattached (#497). Armed by
+    /// `FiberArena::detach_chain`; `None` for chains constructed outside an
+    /// arena (tests). Carries bare slot indices only — never fibers or
+    /// continuations (ISSUE-VM-001 G1); the chain never touches the arena
+    /// directly (Drop can fire at arbitrary Python dealloc points,
+    /// including from another thread).
+    slot_reclaim_queue: Option<Arc<SlotReclaimQueue>>,
 }
 
 impl DetachedFiberChain {
@@ -63,9 +76,17 @@ impl DetachedFiberChain {
             head,
             last_fiber,
             fibers,
+            slot_reclaim_queue: None,
         }
     }
 
+    /// Arm slot reclamation: on drop, any fibers this chain still owns are
+    /// reported to `queue` so the owning arena can return their slots to
+    /// its free list (#497).
+    pub(crate) fn arm_slot_reclaim(&mut self, queue: Arc<SlotReclaimQueue>) {
+        self.slot_reclaim_queue = Some(queue);
+    }
+
     pub fn head(&self) -> FiberId {
         self.head
     }
@@ -78,8 +99,10 @@ impl DetachedFiberChain {
         &self.fibers
     }
 
-    pub fn into_fibers(self) -> Vec<DetachedFiber> {
-        self.fibers
+    pub fn into_fibers(mut self) -> Vec<DetachedFiber> {
+        // Drain in place: `self` then drops with no owned fibers, so the
+        // Drop impl reports nothing — the caller now owns the fibers.
+        std::mem::take(&mut self.fibers)
     }
 
     pub fn set_parent(&mut self, id: FiberId, parent: Option<FiberId>) -> bool {
@@ -267,6 +290,27 @@ impl DetachedFiberChain {
     }
 }
 
+impl Drop for DetachedFiberChain {
+    /// A chain dropped while still owning fibers was abandoned without
+    /// reattachment (abort-style handler, scheduler cancellation, dropped
+    /// parked K). Report the owned slot indices to the arena's reclaim
+    /// queue so they return to the free list instead of stranding as
+    /// vacant-reserved until run end (#497). Consumption paths
+    /// (`into_fibers`, `append`) drain `fibers` first, so they report
+    /// nothing here.
+    fn drop(&mut self) {
+        if self.fibers.is_empty() {
+            return;
+        }
+        let Some(queue) = &self.slot_reclaim_queue else {
+            // Chain never belonged to an arena (direct construction in
+            // tests) — there is no free list to return slots to.
+            return;
+        };
+        queue.report_dropped_slots(self.fibers.iter().map(|entry| entry.id.index()));
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Continuation — the detached fiber chain
 // ---------------------------------------------------------------------------
@@ -452,6 +496,28 @@ impl OwnedControlContinuation {
 // PyK — Python-visible continuation handle (sole owner)
 // ---------------------------------------------------------------------------
 
+/// GC note (#500): PyK deliberately implements NEITHER `__traverse__` nor
+/// `__clear__`, unlike the other Py-holding pyclasses.
+///
+/// - `__clear__` is unsafe by construction: a live PyK is the SOLE owner of
+///   a detached fiber chain (SPEC-VM-021 move-only invariant), and during a
+///   handler dispatch the VM may still recover that chain through a
+///   dispatch-owned `Py<PyK>` handle to reattach it for exception
+///   propagation (#492). Dropping the chain from the GC's `tp_clear` while
+///   such a handle exists would destroy the one-shot ownership invariant
+///   mid-dispatch.
+/// - `__traverse__` alone is not implementable with the current internals:
+///   the Python references inside the chain live behind `PyShared` handles
+///   in fibers → frames → `dyn IRStream` streams → `Value`s, none of which
+///   expose a GC-visit API. Walking them would require threading a visitor
+///   through the whole frame/stream abstraction.
+///
+/// Consequence: the GC treats PyK as an opaque leaf. Cycles that merely
+/// pass through a PyK handle held in a DoExpr field (Resume/Transfer/...)
+/// are still detected via those classes' `__traverse__`; a cycle that
+/// closes through PyK's interior (e.g. a suspended generator captured in
+/// the chain referencing the PyK itself) remains uncollectable until the
+/// continuation is consumed or the PyK is dropped by refcount.
 #[pyclass(name = "K")]
 pub struct PyK {
     continuation: Option<OwnedControlContinuation>,

packages/doeff-vm-core/src/do_ctrl.rs

@@ -68,7 +68,10 @@ pub enum DoCtrl {
 
     /// Install an observer and execute body.
     /// observer: Value::Callable — called synchronously with (effect) on every perform.
-    /// Return value ignored. Original effect always proceeds.
+    /// Return value ignored. If the observer raises, the dispatch is aborted
+    /// and the exception propagates like a handler error — the effect does
+    /// NOT proceed (fail-fast, #506). On success the original effect always
+    /// proceeds unchanged.
     /// body: DoExpr — evaluated under the observer.
     WithObserve { observer: Value, body: Box<DoCtrl> },
 

packages/doeff-vm-core/src/frame.rs

@@ -149,7 +149,19 @@ pub enum EvalReturnContinuation {
         continuation: InterceptorContinuation,
     },
     /// Expand: inner expr evaluated, result must be Value::Stream → push as frame.
-    ExpandReturn,
+    ///
+    /// `handler_k_handle` is Some only when this Expand came from a handler
+    /// dispatch (eval_perform / eval_perform_with_k on the generator-handler
+    /// path, e.g. the @do wrapper's `Expand(Apply(Pure(thunk), []))`). The
+    /// handle owns the perform-site continuation chain for the duration of
+    /// the deferred handler construction; it is consumed on EVERY exit from
+    /// this frame — moved into `Frame::Program.handler_k_handle` on the value
+    /// path, or used to discontinue the perform-site chain on the raise path
+    /// (see step_raise). Storing it here (not in a VM-global slot) ties its
+    /// lifetime to the dispatch that created it.
+    ExpandReturn {
+        handler_k_handle: Option<pyo3::Py<crate::continuation::PyK>>,
+    },
 }
 
 impl EvalReturnContinuation {
@@ -169,7 +181,7 @@ impl EvalReturnContinuation {
             | EvalReturnContinuation::TailResumeReturn
             | EvalReturnContinuation::ReturnToContinuation { .. }
             | EvalReturnContinuation::EvalInScopeReturn { .. }
-            | EvalReturnContinuation::ExpandReturn => None,
+            | EvalReturnContinuation::ExpandReturn { .. } => None,
         }
     }
 }