From a565673f66cf700420582be9db69aa8ac45346ec Mon Sep 17 00:00:00 2001 From: proboscis Date: Fri, 10 Jul 2026 19:02:34 +0900 Subject: [PATCH 1/4] Tie the handler-dispatch k handle's lifetime to its dispatch window (#492) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A wrong-arity @do handler defers its failure: the @do wrapper returns Expand(Apply(Pure(thunk), [])) at call_handler time, and the arity TypeError only fires later inside eval_apply — outside the dispatch's synchronous recovery arm. The VM-global pending_handler_k_handle slot, restored for Expand outcomes, was never cleared on that raise path, so the stale handle was adopted by the NEXT Expand's Program frame: a later unrelated uncaught exception was thrown INTO the long-abandoned perform-site continuation and that continuation's return value was silently substituted as the unrelated program's result (the 2026-07-07 exit-0 incident class, inside the VM itself). Fix: remove the VM-global slot entirely and make the handle a dispatch-local. eval_perform / eval_perform_with_k keep the Py handle in a local; for Expand outcomes it is threaded through the new eval_expand helper into the EvalReturnContinuation::ExpandReturn frame (struct variant), whose two exits both consume it: - value path: step_eval_return moves it onto the resulting Program frame (unchanged recovery semantics for handler-body errors); - raise path: step_raise, on popping an ExpandReturn frame that still owns a handle, recovers the perform-site chain and discontinues it with the exception (OCaml 5: discontinue k exn) — the same routing as the synchronous recovery arm for non-deferred handler failures. If the handler already consumed k, the PyK is empty and the exception propagates to the outer scope. The handle is deliberately NOT carried inside DoCtrl::Expand: do_ctrl.rs documents the instruction set as language-agnostic ("no Python types here"), and the Expand node is user-visible AST constructed all over the Python bridge; the ExpandReturn frame is the VM's own in-flight state for exactly that Expand evaluation, so the Py rides there. As a side effect the dispatch is now re-entrancy-safe (no global slot to clobber) and non-Expand outcomes drop the handle immediately, preserving the stale-backup-leak fix. Invariant checker: the VM-slot root scan is replaced by scanning ExpandReturn frames (live arena fibers and detached chains alike) for owned chains; checker tests root their PyK in an ExpandReturn frame. Regression tests: - Rust: vm_tests test 14 mirrors #492 at the DoCtrl level (deferred Expand(Apply) raising; a later unrelated error must propagate, not be swallowed by the abandoned victim with a substituted Int(666)); the existing stale-backup test now asserts no frame anywhere carries a handle. Verified red pre-fix / green post-fix. - Python: tests/test_handler_exception_catchable.py ports the issue repro — after catching the arity TypeError, an unrelated RuntimeError must escape run() and the victim must not catch it. Verified red on the pre-fix build (DID NOT RAISE) and green post-fix. Fixes #492. Co-Authored-By: Claude Fable 5 --- packages/doeff-vm-core/src/continuation.rs | 4 +- packages/doeff-vm-core/src/frame.rs | 16 +- packages/doeff-vm-core/src/vm.rs | 21 +- packages/doeff-vm-core/src/vm/invariants.rs | 64 ++++-- packages/doeff-vm-core/src/vm/step.rs | 178 +++++++++------ packages/doeff-vm-core/src/vm_tests.rs | 230 +++++++++++++++++++- tests/test_handler_exception_catchable.py | 85 +++++++- 7 files changed, 491 insertions(+), 107 deletions(-) diff --git a/packages/doeff-vm-core/src/continuation.rs b/packages/doeff-vm-core/src/continuation.rs index 568b4d1f..5f72612f 100644 --- a/packages/doeff-vm-core/src/continuation.rs +++ b/packages/doeff-vm-core/src/continuation.rs @@ -16,7 +16,9 @@ //! 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` 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 pyo3::prelude::*; use pyo3::types::PyDict; diff --git a/packages/doeff-vm-core/src/frame.rs b/packages/doeff-vm-core/src/frame.rs index eeeb21f4..9b783187 100644 --- a/packages/doeff-vm-core/src/frame.rs +++ b/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>, + }, } impl EvalReturnContinuation { @@ -169,7 +181,7 @@ impl EvalReturnContinuation { | EvalReturnContinuation::TailResumeReturn | EvalReturnContinuation::ReturnToContinuation { .. } | EvalReturnContinuation::EvalInScopeReturn { .. } - | EvalReturnContinuation::ExpandReturn => None, + | EvalReturnContinuation::ExpandReturn { .. } => None, } } } diff --git a/packages/doeff-vm-core/src/vm.rs b/packages/doeff-vm-core/src/vm.rs index 83cb914c..1b3fe12a 100644 --- a/packages/doeff-vm-core/src/vm.rs +++ b/packages/doeff-vm-core/src/vm.rs @@ -29,26 +29,21 @@ pub struct VM { pub segments: FiberArena, pub var_store: VarStore, pub current_segment: Option, - /// Transient slot used by `eval_perform`/`eval_perform_with_k` to thread - /// a reference to the handler's PyK object through to `push_stream_value`. - /// Set before calling a handler callable; consumed when the resulting - /// Stream frame is pushed (the handle ends up in - /// `Frame::Program.handler_k_handle`). - /// - /// This is a Python handle (Py), NOT a continuation. The chain lives - /// inside the PyK — if the handler raises, the VM borrows this handle, - /// calls `PyK::take()`, and reattaches the recovered chain. - /// Always None outside the handler-dispatch window. - pub pending_handler_k_handle: Option>, } +// NOTE (#492): there is deliberately NO VM-global `pending_handler_k_handle` +// slot. The handler-dispatch PyK handle is owned by the dispatch itself +// (a local in eval_perform/eval_perform_with_k) and, for deferred handler +// construction, by the `EvalReturnContinuation::ExpandReturn` frame — so its +// lifetime is structurally tied to the dispatch window and cannot leak into +// an unrelated later frame. + impl VM { pub fn new() -> Self { VM { segments: FiberArena::new(), var_store: VarStore::new(), current_segment: None, - pending_handler_k_handle: None, } } @@ -56,7 +51,6 @@ impl VM { self.segments.clear(); self.var_store.clear_run_local(); self.current_segment = None; - self.pending_handler_k_handle = None; } pub fn end_active_run_session(&mut self) { @@ -65,7 +59,6 @@ impl VM { self.var_store.clear_run_local(); self.var_store.shrink_run_local_to_fit(); self.current_segment = None; - self.pending_handler_k_handle = None; } pub fn alloc_segment(&mut self, fiber: Fiber) -> FiberId { diff --git a/packages/doeff-vm-core/src/vm/invariants.rs b/packages/doeff-vm-core/src/vm/invariants.rs index fb4ee0c5..23865a35 100644 --- a/packages/doeff-vm-core/src/vm/invariants.rs +++ b/packages/doeff-vm-core/src/vm/invariants.rs @@ -183,17 +183,9 @@ impl VM { fn inv_detached_chains(&self, violations: &mut Vec) -> DetachedView { let mut detached_ids: HashSet = HashSet::new(); - // Roots visible from the VM itself (pending k handle). - if let Some(handle) = &self.pending_handler_k_handle { - self.scan_k_handle( - handle, - "VM.pending_handler_k_handle", - &mut detached_ids, - violations, - ); - } - // Roots inside live arena fibers (frame k handles + scope values). + // There is no VM-global pending-handle slot (#492): every dispatch + // k handle lives either in a Program frame or an ExpandReturn frame. let live_ids: Vec = self.segments.iter().map(|(id, _)| id).collect(); for id in live_ids { if let Some(fiber) = self.segments.get(id) { @@ -272,6 +264,19 @@ impl VM { ); } } + Frame::EvalReturn(cont) => { + if let EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(handle), + } = cont.as_ref() + { + self.scan_k_handle( + handle, + &format!("{origin} (ExpandReturn.handler_k_handle)"), + detached_ids, + violations, + ); + } + } Frame::LexicalScope { bindings, var_overrides, @@ -410,13 +415,23 @@ impl VM { } // Recurse into frames carried by detached fibers that may - // carry their own handler_k_handle. + // carry their own handler_k_handle (Program frames and + // in-flight ExpandReturn frames alike). for frame in &entry.fiber.frames { - if let Frame::Program { - handler_k_handle: Some(nested_handle), - .. - } = frame - { + let nested_handle = match frame { + Frame::Program { + handler_k_handle: Some(handle), + .. + } => Some(handle), + Frame::EvalReturn(cont) => match cont.as_ref() { + EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(handle), + } => Some(handle), + _ => None, + }, + _ => None, + }; + if let Some(nested_handle) = nested_handle { self.scan_k_handle( nested_handle, &format!("{origin} → detached fiber {:?}", entry.id), @@ -479,6 +494,7 @@ impl VM { #[cfg(test)] mod tests { + use crate::frame::{EvalReturnContinuation, Frame}; use crate::ids::Marker; use crate::segment::{Fiber, Handler}; use crate::value::Value; @@ -544,17 +560,24 @@ mod tests { fn properly_detached_chain_holds_invariants() { // Python::attach auto-initializes with the `auto-initialize` feature. let mut vm = VM::new(); + let root = vm.alloc_segment(Fiber::new(None)); let boundary = vm.alloc_segment(Fiber::new(None)); let body = vm.alloc_segment(Fiber::new(Some(boundary))); let chain = vm.segments.detach_chain(body, boundary).unwrap(); let k = crate::continuation::Continuation::from_chain(chain); + // Root the handle where dispatch state lives now (#492): an + // in-flight ExpandReturn frame, not a VM-global slot. pyo3::Python::attach(|py| { let py_k = pyo3::Py::new( py, crate::continuation::PyK::from_continuation(k), ) .unwrap(); - vm.pending_handler_k_handle = Some(py_k); + vm.segments.get_mut(root).unwrap().push_frame(Frame::EvalReturn( + Box::new(EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(py_k), + }), + )); }); assert!(vm.check_invariants().is_ok()); } @@ -564,6 +587,7 @@ mod tests { use crate::continuation::{Continuation, DetachedFiber, DetachedFiberChain}; // Python::attach auto-initializes with the `auto-initialize` feature. let mut vm = VM::new(); + let root = vm.alloc_segment(Fiber::new(None)); // A fiber that is LIVE in the arena... let live = vm.alloc_segment(Fiber::new(None)); // ...and simultaneously claimed by a detached chain — the @@ -583,7 +607,11 @@ mod tests { crate::continuation::PyK::from_continuation(k), ) .unwrap(); - vm.pending_handler_k_handle = Some(py_k); + vm.segments.get_mut(root).unwrap().push_frame(Frame::EvalReturn( + Box::new(EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(py_k), + }), + )); }); let violations = vm.check_invariants().unwrap_err(); assert!(violations diff --git a/packages/doeff-vm-core/src/vm/step.rs b/packages/doeff-vm-core/src/vm/step.rs index e4cf69fa..e8aba8d8 100644 --- a/packages/doeff-vm-core/src/vm/step.rs +++ b/packages/doeff-vm-core/src/vm/step.rs @@ -173,8 +173,30 @@ impl VM { } } _ => { - // Non-program frames can't handle errors — pop and propagate - seg.frames.pop(); + // Non-program frames can't handle errors — pop and propagate. + // + // Exception (#492): an ExpandReturn frame carrying a + // handler-dispatch k handle marks an in-flight deferred + // handler construction (the @do wrapper's + // Expand(Apply(Pure(thunk), [])) shape) whose evaluation + // raised — e.g. an arity TypeError from fn(*args) inside + // eval_apply. The handle is still owned here, so route the + // exception to the perform site's dynamic scope (recover the + // chain and discontinue it), exactly like the synchronous + // recovery arm in eval_perform. Dropping the handle instead + // would strand the perform-site chain; leaving it (the old + // VM-global slot design) let the NEXT Expand adopt it and + // deliver a later unrelated exception into the abandoned + // continuation. + let popped = seg.frames.pop(); + if let Some(Frame::EvalReturn(eval_return)) = popped { + if let EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(handle), + } = *eval_return + { + return self.recover_from_k_handle(handle, error, error_context); + } + } continue_raise(error, error_context) } } @@ -190,23 +212,7 @@ impl VM { DoCtrl::Eval { expr } => continue_eval(*expr, error_context), - DoCtrl::Expand { expr } => { - // Evaluate inner, expect Value::Stream, push as frame - match *expr { - DoCtrl::Pure { value } => self.push_stream_value(value, error_context), - other => { - // Push ExpandReturn frame so we intercept the result - if let Some(seg_id) = self.current_segment { - if let Some(seg) = self.segments.get_mut(seg_id) { - seg.push_frame(Frame::EvalReturn(Box::new( - EvalReturnContinuation::ExpandReturn, - ))); - } - } - continue_eval(other, error_context) - } - } - } + DoCtrl::Expand { expr } => self.eval_expand(*expr, None, error_context), DoCtrl::Apply { f, args } => self.eval_apply(*f, args, error_context), @@ -385,12 +391,50 @@ impl VM { // Helpers // ------------------------------------------------------------------- + /// Evaluate an Expand's inner expr: expect Value::Stream, push as frame. + /// + /// `handler_k_handle` is Some only when the Expand is a handler-dispatch + /// result (the @do wrapper's `Expand(Apply(Pure(thunk), []))` shape from + /// eval_perform/eval_perform_with_k). The handle rides in the ExpandReturn + /// frame so that BOTH exits of the deferred evaluation consume it: the + /// value path moves it onto the resulting Program frame, and the raise + /// path (step_raise popping the ExpandReturn frame) discontinues the + /// perform-site chain with the exception (#492). + fn eval_expand( + &mut self, + expr: DoCtrl, + handler_k_handle: Option>, + error_context: Option>, + ) -> StepResult { + match expr { + DoCtrl::Pure { value } => { + self.push_stream_value(value, handler_k_handle, error_context) + } + other => { + // Push ExpandReturn frame so we intercept the result + if let Some(seg_id) = self.current_segment { + if let Some(seg) = self.segments.get_mut(seg_id) { + seg.push_frame(Frame::EvalReturn(Box::new( + EvalReturnContinuation::ExpandReturn { handler_k_handle }, + ))); + } + } + continue_eval(other, error_context) + } + } + } + /// Push a Value::Stream as a new Program frame on the current fiber. - fn push_stream_value(&mut self, value: Value, error_context: Option>) -> StepResult { - // Consume any k handle stashed by eval_perform/eval_perform_with_k - // so the resulting Program frame can recover the original perform-site - // chain (via PyK.take()) when its stream raises an uncaught exception. - let k_handle = self.pending_handler_k_handle.take(); + /// + /// `k_handle` is the handler-dispatch PyK handle (if any) so the resulting + /// Program frame can recover the original perform-site chain (via + /// PyK.take()) when its stream raises an uncaught exception. + fn push_stream_value( + &mut self, + value: Value, + k_handle: Option>, + error_context: Option>, + ) -> StepResult { match value { Value::Stream(stream) => { if let Some(seg_id) = self.current_segment { @@ -494,49 +538,50 @@ impl VM { // (SPEC-VM-021). We keep a Py handle so that, if the // handler raises before consuming `k`, we can borrow the PyK, // take() the chain, and reattach it for exception propagation - // (OCaml 5 semantics). The handle is stashed in - // `pending_handler_k_handle` so `push_stream_value` can attach - // it to the resulting Program frame. - let k_value = pyo3::Python::attach(|py| { + // (OCaml 5 semantics). The handle is a dispatch-local: every + // arm below consumes or drops it, so it cannot outlive the + // dispatch that created it (#492). + let (k_value, handle) = pyo3::Python::attach(|py| { let py_k = pyo3::Py::new( py, crate::continuation::PyK::from_continuation(k), ) .expect("failed to allocate PyK"); let handle = py_k.clone_ref(py); - self.pending_handler_k_handle = Some(handle); - Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())) + ( + Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())), + handle, + ) }); let outcome = handler_callable.call_handler(vec![effect, k_value]); - // If push_stream_value never ran (because outcome was Err or - // non-Expand), take the handle so it doesn't leak into a future - // handler call. - let leftover_handle = self.pending_handler_k_handle.take(); - match outcome { + Ok(DoCtrl::Expand { expr }) => { + // Deferred handler construction (the @do wrapper returns + // Expand(Apply(Pure(thunk), [])) before fn(*args) runs). + // Thread the handle through eval_expand so it lands on the + // ExpandReturn frame (or the Program frame for Pure) and + // is consumed on both the value and the raise path. + self.eval_expand(*expr, Some(handle), error_context) + } Ok(doctrl) => { - // Only restore the handle for Expand results (which will - // reach push_stream_value). For Pure/other results that - // bypass push_stream_value, dropping the handle prevents - // the stale-backup-leak bug (the chain stays in PyK and - // is freed when the Python K object is GC'd). - if matches!(doctrl, DoCtrl::Expand { .. }) { - self.pending_handler_k_handle = leftover_handle; - } + // Non-Expand result bypasses push_stream_value: drop the + // handle so it can't leak into a future frame (the + // stale-backup-leak bug). The chain stays in PyK and is + // freed when the Python K object is GC'd. + drop(handle); continue_eval(doctrl, error_context) } Err(VMError::UncaughtException { exception }) => { - // Synchronous Python exception from call_handler itself - // (the @do wrapper's Expand construction raised). Recover - // the chain from the PyK handle. - match leftover_handle { - Some(handle) => { - self.recover_from_k_handle(handle, exception, error_context) - } - None => continue_raise(exception, error_context), - } + // Synchronous Python exception from call_handler itself. + // The handle is still owned here, so route the exception + // to the perform site's dynamic scope: recover the chain + // from the PyK handle and raise into it (OCaml 5: + // discontinue k exn). If the handler already consumed k, + // the PyK is empty and the raise lands on current_segment + // (the outer scope). + self.recover_from_k_handle(handle, exception, error_context) } Err(err) => error_result(err, error_context), } @@ -738,33 +783,32 @@ impl VM { if handler_callable.is_generator_handler() { // Generator handler path — see eval_perform for rationale. - let k_value = pyo3::Python::attach(|py| { + let (k_value, handle) = pyo3::Python::attach(|py| { let py_k = pyo3::Py::new( py, crate::continuation::PyK::from_continuation(k), ) .expect("failed to allocate PyK"); let handle = py_k.clone_ref(py); - self.pending_handler_k_handle = Some(handle); - Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())) + ( + Value::Opaque(crate::py_shared::PyShared::new(py_k.into_any())), + handle, + ) }); let outcome = handler_callable.call_handler(vec![effect, k_value]); - let leftover_handle = self.pending_handler_k_handle.take(); match outcome { + Ok(DoCtrl::Expand { expr }) => { + self.eval_expand(*expr, Some(handle), error_context) + } Ok(doctrl) => { - if matches!(doctrl, DoCtrl::Expand { .. }) { - self.pending_handler_k_handle = leftover_handle; - } + drop(handle); continue_eval(doctrl, error_context) } - Err(VMError::UncaughtException { exception }) => match leftover_handle { - Some(handle) => { - self.recover_from_k_handle(handle, exception, error_context) - } - None => continue_raise(exception, error_context), - }, + Err(VMError::UncaughtException { exception }) => { + self.recover_from_k_handle(handle, exception, error_context) + } Err(err) => error_result(err, error_context), } } else { @@ -849,7 +893,9 @@ impl VM { } } EvalReturnContinuation::TailResumeReturn => continue_send(value, error_context), - EvalReturnContinuation::ExpandReturn => self.push_stream_value(value, error_context), + EvalReturnContinuation::ExpandReturn { handler_k_handle } => { + self.push_stream_value(value, handler_k_handle, error_context) + } _ => { // Other EvalReturn variants — TODO continue_send(value, error_context) diff --git a/packages/doeff-vm-core/src/vm_tests.rs b/packages/doeff-vm-core/src/vm_tests.rs index 673c3a61..feb03df0 100644 --- a/packages/doeff-vm-core/src/vm_tests.rs +++ b/packages/doeff-vm-core/src/vm_tests.rs @@ -1343,13 +1343,233 @@ mod tests { Err(err) => panic!("expected Ok, got error: {:?}", err), } - // Critical assertion: pending_handler_k_handle must be None. - // With the stale-backup bug, the handle would persist and - // attach to the next unrelated Program frame. + // Critical assertion: no dispatch k handle may survive anywhere in + // VM state. The VM-global pending_handler_k_handle slot was removed + // (#492) — the handle is dispatch-local and, for non-Expand results, + // dropped inside eval_perform — so the only places a handle can live + // are Program / ExpandReturn frames. Scan them all. assert!( - vm.pending_handler_k_handle.is_none(), - "stale-backup-leak: pending_handler_k_handle must be None \ + no_leaked_k_handle(&vm), + "stale-backup-leak: no frame may carry a handler k handle \ after a generator handler returns a non-Expand DoCtrl" ); } + + /// True if no live arena fiber carries a handler-dispatch k handle + /// (Program.handler_k_handle or ExpandReturn.handler_k_handle). + fn no_leaked_k_handle(vm: &VM) -> bool { + vm.segments.iter().all(|(_, fiber)| { + fiber.frames.iter().all(|frame| match frame { + Frame::Program { + handler_k_handle, .. + } => handler_k_handle.is_none(), + Frame::EvalReturn(cont) => !matches!( + cont.as_ref(), + crate::frame::EvalReturnContinuation::ExpandReturn { + handler_k_handle: Some(_), + } + ), + _ => true, + }) + }) + } + + // ----------------------------------------------------------------------- + // Test 14 (#492): a deferred handler-construction failure (the @do + // wrapper's Expand(Apply(Pure(thunk), [])) shape, where the thunk raises + // an arity TypeError inside eval_apply) must not leave a stale k handle + // behind. Pre-fix, the stale handle was adopted by the NEXT Expand frame, + // and a later unrelated exception was delivered INTO the abandoned + // perform-site continuation, whose return value was then substituted as + // the unrelated program's result. + // ----------------------------------------------------------------------- + + #[test] + fn test_deferred_expand_apply_error_routes_to_perform_site_and_leaves_no_stale_handle() { + use crate::continuation::PyK; + + /// Callable that raises when called — models fn(*args) blowing up + /// with an arity TypeError inside the @do wrapper's thunk. + #[derive(Debug)] + struct RaisingThunk; + + impl Callable for RaisingThunk { + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn call(&self, _args: Vec) -> Result { + Err(VMError::UncaughtException { + exception: Value::String("arity TypeError".into()), + }) + } + } + + /// Generator-like handler that returns the deferred-construction + /// shape WITHOUT touching k: Expand(Apply(Pure(RaisingThunk), [])). + /// The k stays inside the PyK the VM allocated — exactly the @do + /// wrong-arity situation from #492. + #[derive(Debug)] + struct DeferredCrashingHandler; + + impl Callable for DeferredCrashingHandler { + fn as_any(&self) -> &dyn std::any::Any { + self + } + fn call(&self, _args: Vec) -> Result { + Err(VMError::internal("use call_handler")) + } + fn is_generator_handler(&self) -> bool { + true + } + fn call_handler(&self, args: Vec) -> Result { + // Sanity: k arrives as Opaque(PyK) on the generator path. + pyo3::Python::attach(|py| match &args[1] { + Value::Opaque(obj) => { + obj.bind(py).cast::().expect("expected PyK"); + } + other => panic!("expected Opaque(PyK), got {:?}", other), + }); + Ok(DoCtrl::Expand { + expr: Box::new(DoCtrl::Apply { + f: Box::new(DoCtrl::Pure { + value: Value::Callable(Arc::new(RaisingThunk) as CallableRef), + }), + args: vec![], + }), + }) + } + } + + /// Victim body: performs, and its throw() arm swallows ONLY the + /// unrelated crash (like the Python repro's `except RuntimeError`), + /// returning a recognizable substituted value. The construction + /// TypeError — correctly discontinued into the victim's chain — + /// propagates through. If the stale-handle bug is present, the later + /// unrelated crash lands HERE and Int(666) becomes the unrelated + /// program's "result". + #[derive(Debug)] + struct VictimStream { + state: u8, + } + + impl IRStream for VictimStream { + fn resume(&mut self, value: Value) -> StreamStep { + match self.state { + 0 => { + self.state = 1; + StreamStep::Instruction(DoCtrl::Perform { + effect: Value::String("ping".into()), + }) + } + _ => StreamStep::Done(value), + } + } + fn throw(&mut self, e: Value) -> StreamStep { + match &e { + Value::String(s) if s == "unrelated crash" => { + StreamStep::Done(Value::Int(666)) + } + _ => StreamStep::Error(e), + } + } + } + + /// Crasher body: raises an unrelated error, uncaught. + #[derive(Debug)] + struct CrasherStream; + + impl IRStream for CrasherStream { + fn resume(&mut self, _value: Value) -> StreamStep { + StreamStep::Error(Value::String("unrelated crash".into())) + } + fn throw(&mut self, e: Value) -> StreamStep { + StreamStep::Error(e) + } + } + + /// Main: installs the bad handler around the victim, CATCHES the + /// construction error (throw → continues), then runs the crasher as + /// an unrelated subprogram. The crasher's error must propagate out. + #[derive(Debug)] + struct MainStream { + state: u8, + h: Option, + b: Option>, + caught: Option, + } + + impl IRStream for MainStream { + fn resume(&mut self, value: Value) -> StreamStep { + match self.state { + 0 => { + self.state = 1; + StreamStep::Instruction(DoCtrl::WithHandler { + handler: self.h.take().unwrap(), + body: self.b.take().unwrap(), + }) + } + 1 => { + // WithHandler returned normally — the construction + // error was NOT raised. That would itself be a bug. + StreamStep::Error(Value::String( + "expected construction TypeError, got value".into(), + )) + } + 2 => { + self.state = 3; + StreamStep::Instruction(DoCtrl::Expand { + expr: Box::new(DoCtrl::Pure { + value: Value::Stream(IRStreamRef::new( + Box::new(CrasherStream) as Box, + )), + }), + }) + } + _ => StreamStep::Done(value), + } + } + fn throw(&mut self, e: Value) -> StreamStep { + if self.state == 1 { + // Caught the handler-construction error — proceed to the + // unrelated crasher, like the Python repro's try/except. + self.state = 2; + self.caught = Some(e); + self.resume(Value::Unit) + } else { + StreamStep::Error(e) + } + } + } + + let mut vm = setup_vm_with_stream(MainStream { + state: 0, + h: Some(Value::Callable( + Arc::new(DeferredCrashingHandler) as CallableRef, + )), + b: Some(expand_stream(VictimStream { state: 0 })), + caught: None, + }); + + let result = run_to_completion(&mut vm); + + // The unrelated crash must propagate out as an uncaught error — + // NOT be swallowed by the abandoned victim continuation (which + // would substitute Ok(Int(666)) here). + match result { + Err(VMError::UncaughtException { exception }) => match exception { + Value::String(s) => assert_eq!(s, "unrelated crash"), + other => panic!("expected String exception, got {:?}", other), + }, + Ok(other) => panic!( + "unrelated crash was swallowed; wrong-scope substituted result: {:?}", + other + ), + Err(err) => panic!("expected UncaughtException, got {:?}", err), + } + + assert!( + no_leaked_k_handle(&vm), + "stale k handle survived the deferred handler-construction failure" + ); + } } diff --git a/tests/test_handler_exception_catchable.py b/tests/test_handler_exception_catchable.py index 7ce1821b..39c6a826 100644 --- a/tests/test_handler_exception_catchable.py +++ b/tests/test_handler_exception_catchable.py @@ -17,8 +17,9 @@ from dataclasses import dataclass +import pytest from doeff_core_effects.handlers import try_handler -from doeff_vm import Err +from doeff_vm import Err, WithHandler from doeff import ( EffectBase, @@ -242,3 +243,85 @@ def program(): assert isinstance(result, Err), f"Expected Err, got {result!r}" assert isinstance(result.error, RuntimeError) assert "outer handler failed" in str(result.error) + + +# --------------------------------------------------------------------------- +# Tests (#492): deferred @do handler-construction failure must not leave a +# stale perform-site continuation behind. +# +# A wrong-arity @do handler matters because the @do wrapper returns +# Expand(Apply(Pure(thunk), [])) immediately; the arity TypeError only fires +# later, inside eval_apply, when the thunk calls fn(*args) — i.e. outside the +# synchronous dispatch recovery. Pre-fix, the perform-site k handle leaked +# into a VM-global slot, was adopted by the NEXT @do call's Expand frame, and +# a later unrelated exception was delivered INTO the long-abandoned victim +# continuation — whose return value was then silently substituted as the +# unrelated program's result (the 2026-07-07 exit-0 incident class). +# --------------------------------------------------------------------------- + + +@do +def _do_wrong_arity_handler(effect): # wrong arity: handlers take (effect, k) + yield None + + +def test_do_handler_wrong_arity_type_error_catchable(): + """The deferred arity TypeError routes to the perform site's dynamic scope + and is catchable around WithHandler — same as the plain-handler case.""" + + @do + def victim(): + yield Ping(label="x") + return "victim: no exception" + + @do + def program(): + try: + result = yield WithHandler(_do_wrong_arity_handler, victim()) + return result + except TypeError as e: + return f"caught: {e}" + + result = run(program()) + assert result.startswith("caught: "), f"Expected caught TypeError, got {result!r}" + assert "positional argument" in result + + +def test_stale_handle_not_adopted_by_next_program(): + """#492 repro: after catching the arity TypeError, a later unrelated + uncaught RuntimeError must propagate out of run() — NOT be delivered into + the abandoned victim continuation (which would substitute the victim's + return value as the crasher's result).""" + + caught_type_errors: list[TypeError] = [] + + @do + def victim(): + try: + yield Ping(label="x") + return "victim: no exception" + except RuntimeError as e: + return f"VICTIM CAUGHT WRONG-SCOPE EXCEPTION: {e}" + + @do + def crasher(): + if True: + raise RuntimeError("crash in unrelated subprogram") + yield # unreachable, makes it a generator + + @do + def main(): + try: + yield WithHandler(_do_wrong_arity_handler, victim()) + except TypeError as e: + caught_type_errors.append(e) + r = yield crasher() + return ("crasher returned", r) + + with pytest.raises(RuntimeError, match="crash in unrelated subprogram"): + run(main()) + + assert len(caught_type_errors) == 1, ( + "the deferred handler-construction arity TypeError must surface " + f"exactly once at the dispatch's dynamic scope, got {caught_type_errors!r}" + ) From 3eebe8360be132acdd2e464745f6ce310792c0bc Mon Sep 17 00:00:00 2001 From: proboscis Date: Fri, 10 Jul 2026 19:25:49 +0900 Subject: [PATCH 2/4] Implement the PyO3 GC protocol on Py-holding pyclasses so cycles through VM objects are collectable (#500) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit No doeff_vm pyclass participated in CPython's cycle GC, so any reference cycle through a program node, effect, callable wrapper, stream, or result was permanently uncollectable — every accidental cycle pinned its whole reachable subgraph for the life of the process. Mechanism: - __traverse__ visiting every Py/Py field is added to Pure, Perform, Resume, Transfer, Apply, Expand, Pass, WithHandler, ResumeThrow, TransferThrow, WithObserve, GetTraceback, GetHandlers, GetBoundaries, TailEval (do_expr.rs), Callable/PythonCallable and IRStream/PyIRStream (python_generator_stream.rs), and Ok/Err (result.rs). Defining __traverse__ sets Py_TPFLAGS_HAVE_GC, so the collector can now see through these objects. - __clear__ is implemented only on the non-frozen classes (PythonCallable, PyIRStream), replacing the held reference with None. The DoExpr and Ok/Err classes are `frozen` (no &mut self in pyo3), so their references cannot be dropped in-place; traverse-only is sound because field cycles cannot be constructed among frozen nodes alone — every real cycle routes through at least one mutable Python object whose tp_clear breaks it once traversal has made the cycle visible. - EffectBase drops its pyo3 `dict` slot. pyo3 0.28 cannot visit a pyclass dict slot from __traverse__ (the method receives only &self), so a base-owned __dict__ was invisible to the GC. Without the base slot, Python subclasses (every real effect) get a CPython-managed __dict__ that subtype_traverse/subtype_clear handle natively, making cycles through effect attributes collectable. Direct EffectBase() instances no longer accept attribute assignment; nothing in the repo does that (effects are always subclasses). - PyK is deliberately EXCLUDED (documented in continuation.rs): a live PyK is the sole owner of a detached fiber chain (SPEC-VM-021), and a GC-driven __clear__ could drop the chain while a dispatch-owned Py handle may still recover it for exception propagation (#492), breaking the one-shot ownership invariant mid-dispatch. A traverse-only impl is not currently implementable either: the chain's Python refs live behind PyShared in fibers/frames/dyn IRStream values with no GC-visit API. PyK stays an opaque GC leaf. Known residual limitation (documented in do_expr.rs): the pyo3 `dict` slot kept on DoExpr classes for defp metadata (__doeff_body__ etc.) is still not traversable in pyo3 0.28, so a cycle routed exclusively through a program node's instance __dict__ remains invisible. Regression tests (TestGcCycleCollection in test_pyvm.py) port the issue repro: cycles through Pure(o), through EffectBase-subclass attributes, between two EffectBase instances, and through WithHandler/Apply must be collected (weakref dies); pure-Python control included. Verified red on the unfixed build, green after. Co-Authored-By: Claude Fable 5 --- packages/doeff-vm-core/src/continuation.rs | 22 +++++ packages/doeff-vm/src/do_expr.rs | 88 +++++++++++++++++ .../doeff-vm/src/python_generator_stream.rs | 35 ++++++- packages/doeff-vm/src/result.rs | 15 +++ packages/doeff-vm/tests/test_pyvm.py | 95 +++++++++++++++++++ 5 files changed, 254 insertions(+), 1 deletion(-) diff --git a/packages/doeff-vm-core/src/continuation.rs b/packages/doeff-vm-core/src/continuation.rs index 5f72612f..38149c3b 100644 --- a/packages/doeff-vm-core/src/continuation.rs +++ b/packages/doeff-vm-core/src/continuation.rs @@ -454,6 +454,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` 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, diff --git a/packages/doeff-vm/src/do_expr.rs b/packages/doeff-vm/src/do_expr.rs index cf08b0f3..5bd8def8 100644 --- a/packages/doeff-vm/src/do_expr.rs +++ b/packages/doeff-vm/src/do_expr.rs @@ -2,9 +2,29 @@ //! //! These replace the plain Python classes in `doeff/program.py`. //! The VM classifies them via `downcast` (not tag-based `getattr`). +//! +//! ## GC integration (#500) +//! +//! Every class that holds `Py` / `Py` fields implements +//! `__traverse__` so CPython's cycle collector can see through it — +//! without it, any reference cycle through a program node is permanently +//! uncollectable. `__clear__` is deliberately NOT implemented: these +//! classes are `frozen` (no `&mut self` access, required by the VM's +//! immutable-program invariant), so their field references cannot be +//! dropped in-place. That is sound for collection: field cycles cannot be +//! constructed among frozen nodes alone (fields are set once at +//! construction), so every real cycle routes through at least one mutable +//! Python object (instance `__dict__`, list, generator frame, ...) whose +//! `tp_clear` breaks the cycle once `__traverse__` has made it visible. +//! +//! Known limitation: the pyo3 `dict` slot (`#[pyclass(dict)]`, used by +//! defp for `__doeff_body__` metadata) is NOT reachable from +//! `__traverse__` in pyo3 0.28, so a cycle routed exclusively through a +//! program node's instance `__dict__` is still invisible to the GC. use doeff_vm_core::continuation::PyK; use pyo3::prelude::*; +use pyo3::pyclass::{PyTraverseError, PyVisit}; /// Pure(value) — return a value immediately. #[pyclass(name = "Pure", frozen, dict, module = "doeff_vm.doeff_vm")] @@ -29,6 +49,10 @@ impl PyPure { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.value.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.value) + } } /// Perform(effect) — perform an effect (trigger handler lookup). @@ -54,6 +78,10 @@ impl PyPerform { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.effect.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.effect) + } } /// Resume(k, value) — resume continuation with value (non-tail, handler stays alive). @@ -78,6 +106,11 @@ impl PyResume { fn __repr__(&self) -> &'static str { "Resume(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.value) + } } /// Transfer(k, value) — resume continuation with value (tail, handler done). @@ -102,6 +135,11 @@ impl PyTransfer { fn __repr__(&self) -> &'static str { "Transfer(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.value) + } } /// Apply(f, args) — call f(args). @@ -128,6 +166,11 @@ impl PyApply { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.f.clone_ref(py), self.args.clone_ref(py)))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.f)?; + visit.call(&self.args) + } } /// Expand(expr) — evaluate inner expr to Stream, then run it. @@ -152,6 +195,10 @@ impl PyExpand { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.expr.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.expr) + } } /// Pass(effect, k) — handler doesn't handle, forward to outer. @@ -176,6 +223,11 @@ impl PyPass { fn __repr__(&self) -> &'static str { "Pass(effect, k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.effect)?; + visit.call(&self.continuation) + } } /// WithHandler(handler, body) — install handler and run body under it. @@ -216,6 +268,11 @@ impl PyWithHandler { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.handler.clone_ref(py), self.body.clone_ref(py)))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.handler)?; + visit.call(&self.body) + } } /// ResumeThrow(k, exception) — throw exception into continuation (non-tail). @@ -240,6 +297,11 @@ impl PyResumeThrow { fn __repr__(&self) -> &'static str { "ResumeThrow(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.exception) + } } /// TransferThrow(k, exception) — throw exception into continuation (tail). @@ -264,6 +326,11 @@ impl PyTransferThrow { fn __repr__(&self) -> &'static str { "TransferThrow(k, ...)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation)?; + visit.call(&self.exception) + } } /// WithObserve(observer, body) — install observer and run body under it. @@ -290,6 +357,11 @@ impl PyWithObserve { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.observer.clone_ref(py), self.body.clone_ref(py)))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.observer)?; + visit.call(&self.body) + } } /// GetTraceback(k) — query traceback from continuation without consuming it. @@ -309,6 +381,10 @@ impl PyGetTraceback { fn __repr__(&self) -> &'static str { "GetTraceback(k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation) + } } /// GetExecutionContext() — get current execution context. @@ -354,6 +430,10 @@ impl PyGetHandlers { fn __repr__(&self) -> &'static str { "GetHandlers(k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation) + } } /// GetBoundaries(k) — extract the interleaved handler/observer boundary @@ -380,6 +460,10 @@ impl PyGetBoundaries { fn __repr__(&self) -> &'static str { "GetBoundaries(k)" } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.continuation) + } } /// GetOuterHandlers — extract handlers installed ABOVE the current handler. @@ -426,4 +510,8 @@ impl PyTailEval { let e = self.expr.bind(py).repr()?; Ok(format!("TailEval({})", e)) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.expr) + } } diff --git a/packages/doeff-vm/src/python_generator_stream.rs b/packages/doeff-vm/src/python_generator_stream.rs index fbfd34fc..6cecf081 100644 --- a/packages/doeff-vm/src/python_generator_stream.rs +++ b/packages/doeff-vm/src/python_generator_stream.rs @@ -7,6 +7,7 @@ use pyo3::exceptions::PyStopIteration; use pyo3::prelude::*; +use pyo3::pyclass::{PyTraverseError, PyVisit}; use pyo3::types::PyString; use doeff_vm_core::do_ctrl::DoCtrl; @@ -18,7 +19,18 @@ use doeff_vm_core::value::Value; /// The Rust side uses `is_instance_of::()` for classification. /// /// Yielding an EffectBase from a generator is implicitly treated as Perform(effect). -#[pyclass(name = "EffectBase", subclass, dict, module = "doeff_vm.doeff_vm")] +/// +/// GC note (#500): EffectBase deliberately does NOT declare a pyo3 `dict` +/// slot. pyo3 0.28 cannot visit a pyclass dict slot from `__traverse__` +/// (the method only receives `&self`), so a base-owned `__dict__` would be +/// invisible to the cycle collector — any reference cycle through an +/// effect's attributes would be permanently uncollectable. Without a base +/// dict slot, Python subclasses (every real effect is a subclass) get a +/// CPython-managed `__dict__` which `subtype_traverse`/`subtype_clear` +/// handle natively, making cycles through effect attributes collectable. +/// Consequence: direct `EffectBase()` instances have no `__dict__` and +/// reject attribute assignment — effects must be subclasses (they are). +#[pyclass(name = "EffectBase", subclass, module = "doeff_vm.doeff_vm")] #[derive(Debug)] pub struct PyEffectBase; @@ -70,6 +82,17 @@ impl PythonCallable { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.callable.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.callable) + } + + fn __clear__(&mut self, py: Python<'_>) { + // Break cycles through the wrapped callable. The object is GC + // garbage at this point; any buggy post-clear use fails loudly + // ("'NoneType' object is not callable"). + self.callable = py.None(); + } } impl doeff_vm_core::value::Callable for PythonCallable { @@ -161,6 +184,16 @@ impl PyIRStream { tail_resume_lines: tail_resume_lines.unwrap_or_default(), } } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.generator) + } + + fn __clear__(&mut self, py: Python<'_>) { + // Break cycles through the wrapped generator. The object is GC + // garbage at this point; any buggy post-clear use fails loudly. + self.generator = py.None(); + } } /// A Python generator wrapped as an IRStream. diff --git a/packages/doeff-vm/src/result.rs b/packages/doeff-vm/src/result.rs index 0fc8780a..257925e4 100644 --- a/packages/doeff-vm/src/result.rs +++ b/packages/doeff-vm/src/result.rs @@ -1,4 +1,10 @@ use pyo3::prelude::*; +use pyo3::pyclass::{PyTraverseError, PyVisit}; + +// GC note (#500): both result classes implement `__traverse__` so the cycle +// collector can see through them. They are `frozen`, so `__clear__` is not +// possible (no `&mut self`); see the module doc of `do_expr.rs` for why +// traverse-only is sound for frozen value holders. #[pyclass(frozen, name = "Ok", module = "doeff_vm.doeff_vm")] pub struct PyResultOk { @@ -43,6 +49,10 @@ impl PyResultOk { let cls = py.get_type::().into_any().unbind(); Ok((cls, (self.value.clone_ref(py),))) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.value) + } } #[pyclass(frozen, name = "Err", module = "doeff_vm.doeff_vm")] @@ -104,4 +114,9 @@ impl PyResultErr { ), )) } + + fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> { + visit.call(&self.error)?; + visit.call(&self.captured_traceback) + } } diff --git a/packages/doeff-vm/tests/test_pyvm.py b/packages/doeff-vm/tests/test_pyvm.py index e7a5054a..d36bf2fd 100644 --- a/packages/doeff-vm/tests/test_pyvm.py +++ b/packages/doeff-vm/tests/test_pyvm.py @@ -228,3 +228,98 @@ def test_vm_live_counts_return_to_baseline_after_run() -> None: assert run(simple_program()) == 42 assert doeff_vm.vm_live_counts() == before + + +class TestGcCycleCollection: + """Regression tests for #500: Py-holding pyclasses must implement the GC + protocol (__traverse__) so reference cycles through doeff_vm objects are + collectable. Before the fix, any cycle through Pure(...) or through an + EffectBase instance's attributes was permanently uncollectable. + """ + + @staticmethod + def _collect_and_check(ref) -> bool: + import gc + + for _ in range(3): + gc.collect() + return ref() is None + + def test_pure_python_control_cycle_is_collected(self) -> None: + """Sanity control: a pure-Python two-object cycle is collectable.""" + import weakref + + class Holder: + pass + + a, b = Holder(), Holder() + a.x = b + b.x = a + ref = weakref.ref(a) + del a, b + assert self._collect_and_check(ref) + + def test_cycle_through_pure_is_collected(self) -> None: + """o -> Pure(o) -> o must be collected (issue #500 repro).""" + import weakref + + class Holder: + pass + + o = Holder() + p = doeff_vm.Pure(o) + o.cycle = p + ref = weakref.ref(o) + del o, p + assert self._collect_and_check(ref) + + def test_cycle_through_effect_base_instance_is_collected(self) -> None: + """A cycle through an EffectBase subclass instance's attributes + must be collected (issue #500 repro, EffectBase variant).""" + import weakref + + class Holder: + pass + + class CycleEffect(doeff_vm.EffectBase): + pass + + e = CycleEffect() + o = Holder() + e.o = o + o.e = e + ref = weakref.ref(o) + del e, o + assert self._collect_and_check(ref) + + def test_cycle_between_two_effect_base_instances_is_collected(self) -> None: + """Two EffectBase instances referencing each other must be collected.""" + import weakref + + class EffA(doeff_vm.EffectBase): + pass + + class EffB(doeff_vm.EffectBase): + pass + + a, b = EffA(), EffB() + a.x = b + b.x = a + ref = weakref.ref(a) + del a, b + assert self._collect_and_check(ref) + + def test_cycle_through_with_handler_and_apply_is_collected(self) -> None: + """Cycles through composite nodes (WithHandler/Apply/Perform fields) + must be visible to the GC via __traverse__.""" + import weakref + + class Holder: + pass + + o = Holder() + node = doeff_vm.WithHandler(lambda _e, _k: None, doeff_vm.Apply(print, [o])) + o.cycle = node + ref = weakref.ref(o) + del o, node + assert self._collect_and_check(ref) From eda61eb2b4e5d67136eb275ab8fbc5891be0617e Mon Sep 17 00:00:00 2001 From: proboscis Date: Fri, 10 Jul 2026 19:26:04 +0900 Subject: [PATCH 3/4] Fail fast on observer exceptions instead of silently swallowing them (#506) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit call_all_observers discarded observer errors (`let _ = observer.call(...)`), so an observer that raised was silently dropped: the effect proceeded and the exception never surfaced. Observers are exactly where tracing/slog/OTel layers hang — a broken audit layer in a long-lived process died silently and stayed dead, against the fail-fast doctrine downstream repos rely on. Mechanism: call_all_observers now returns Result<(), VMError>; the first failing observer aborts the walk. eval_perform propagates the error before the handler is looked up — an UncaughtException is raised at the perform site via continue_raise (identical to a synchronous handler exception, so try/except around the yield sees it and an uncaught one propagates out of run()); other VMErrors go through error_result. The effect is NOT delivered to any handler. The DoCtrl::WithObserve doc comment now states the fail-fast contract explicitly (it previously promised only "return value ignored"). Existing tests relying on lenient observers: none found — a repo-wide survey of observer tests (tests/test_intercept.py, tests/test_with_observe_visibility.py, tests/test_spawn_observer_inheritance.py, packages/doeff-vm*/tests) shows no test raises from an observer, and the full suite (964 passed) is unchanged, so no deliberate updates were needed. Regression tests port the issue repro: a raising observer must make run() raise that exception instead of returning the handled value, and the exception must be catchable at the perform site. Verified red on the unfixed build, green after. Co-Authored-By: Claude Fable 5 --- packages/doeff-vm-core/src/do_ctrl.rs | 5 ++- packages/doeff-vm-core/src/vm/step.rs | 29 ++++++++++++-- tests/test_with_observe_visibility.py | 54 +++++++++++++++++++++++++++ 3 files changed, 83 insertions(+), 5 deletions(-) diff --git a/packages/doeff-vm-core/src/do_ctrl.rs b/packages/doeff-vm-core/src/do_ctrl.rs index ba9458c0..965dbfc8 100644 --- a/packages/doeff-vm-core/src/do_ctrl.rs +++ b/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 }, diff --git a/packages/doeff-vm-core/src/vm/step.rs b/packages/doeff-vm-core/src/vm/step.rs index e8aba8d8..264d4312 100644 --- a/packages/doeff-vm-core/src/vm/step.rs +++ b/packages/doeff-vm-core/src/vm/step.rs @@ -520,8 +520,22 @@ impl VM { } }; - // 1. Call ALL observers in the chain (synchronous, return value ignored) - self.call_all_observers(current, &effect); + // 1. Call ALL observers in the chain (synchronous, return value + // ignored). An observer exception FAILS FAST: it aborts the + // dispatch and propagates like a handler error — the effect is + // NOT delivered to any handler (#506). A dead tracing/audit + // layer must be loud, not silently dropped. + if let Err(err) = self.call_all_observers(current, &effect) { + return match err { + VMError::UncaughtException { exception } => { + // Python exception from the observer — raise it at the + // perform site so try/except around the yield sees it, + // exactly like a synchronous handler exception. + continue_raise(exception, error_context) + } + err => error_result(err, error_context), + }; + } // 2. Proceed to handler let result = match self.perform_effect(&effect) { @@ -634,7 +648,13 @@ impl VM { } /// Walk the entire chain and call all observers synchronously. - fn call_all_observers(&self, start: FiberId, effect: &Value) { + /// + /// Observer RETURN VALUES are ignored, but observer ERRORS are not: + /// the first failing observer aborts the walk and its error is + /// propagated by the caller like a handler error (fail-fast, #506). + /// Observers outside the failing one (further up the chain) are not + /// called for this effect. + fn call_all_observers(&self, start: FiberId, effect: &Value) -> Result<(), VMError> { let mut cursor = Some(start); while let Some(fid) = cursor { let Some(seg) = self.segments.get(fid) else { @@ -642,11 +662,12 @@ impl VM { }; if seg.is_intercept_boundary() { if let Some(observer) = seg.intercept_handler().cloned() { - let _ = observer.call(vec![effect.clone()]); + observer.call(vec![effect.clone()])?; } } cursor = seg.parent; } + Ok(()) } /// Evaluate WithObserve: install observer boundary, create body fiber, evaluate body. diff --git a/tests/test_with_observe_visibility.py b/tests/test_with_observe_visibility.py index 670805d8..a9e97884 100644 --- a/tests/test_with_observe_visibility.py +++ b/tests/test_with_observe_visibility.py @@ -91,3 +91,57 @@ def prog(): tell_msgs = [o for o in observed if "tell from handler" in o] assert slog_msgs, f"Observer missed handler body slog. Observed: {observed}" assert tell_msgs, f"Observer missed handler body Tell. Observed: {observed}" + + +def test_observer_exception_fails_fast(): + """Regression test for #506: an exception raised by an observer must + abort the dispatch and propagate like a handler error — run() raises + instead of returning the handled value. A dead tracing/audit layer must + be loud, not silently swallowed. + """ + import pytest + + class Ping(EffectBase): + pass + + def exploding_observer(effect): + raise RuntimeError("observer exploded") + + @do + def h(effect, k): + return (yield Resume(k, "handled")) + + @do + def body(): + return (yield Ping()) + + program = WithObserve(VMCallable(exploding_observer), _program_handler(h)(body())) + with pytest.raises(RuntimeError, match="observer exploded"): + run(program) + + +def test_observer_exception_is_catchable_at_perform_site(): + """#506: the observer exception is raised at the perform site, so a + try/except around the yield can catch it (same semantics as a + synchronous handler exception).""" + + class Ping(EffectBase): + pass + + def exploding_observer(effect): + raise RuntimeError("observer exploded") + + @do + def h(effect, k): + return (yield Resume(k, "handled")) + + @do + def body(): + try: + yield Ping() + except RuntimeError as exc: + return f"caught: {exc}" + return "not raised" + + program = WithObserve(VMCallable(exploding_observer), _program_handler(h)(body())) + assert run(program) == "caught: observer exploded" From c4fefe61d9b3f3ef3aa48d9c55526421b9296377 Mon Sep 17 00:00:00 2001 From: proboscis Date: Fri, 10 Jul 2026 19:48:53 +0900 Subject: [PATCH 4/4] Reclaim arena slots when a detached fiber chain is dropped unresumed (#497) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dropping a continuation without resuming it (abort-style handler, scheduler cancellation, abandoned parked K) left its arena slots vacant-reserved forever within the run: detach_chain empties the slots, attach_chain is the only path that refills them, and free() is never called for dropped chains. Every abandoned dispatch stranded ~2 slots, growing the fiber slot vector unboundedly (measured: head fiber index 40000 after 20k aborts) and dragging ~528 B/abort of RSS that persisted across runs via heap fragmentation of the regrown Vec>. Mechanism: the arena owns a SlotReclaimQueue (Arc>>); detach_chain arms each DetachedFiberChain with a handle to it. The chain's new Drop impl reports the slot indices it still owns — its fibers are destroyed normally by the same Drop; only bare indices flow back, never fibers or continuations, so SPEC-VM-021 single-owner semantics and the G1 fiber-ownership guards are preserved (this is the arena analogue of OCaml 5's "k dropped => fibers freed by GC"). into_fibers/append drain the fiber vec first and therefore report nothing. alloc() drains the queue into the free list before allocating; clear() swaps in a fresh queue so chains outliving a run session cannot poison the next session's free list. Under invariant-checks, reclaim panics if a reported slot is not vacant-reserved. Measured after the fix: abort-loop head fiber index stays at 2 (was 2*N), and RSS plateaus at 0 KB/run growth by run 5 (was +9.5-10 MB per 20k-abort run, no plateau) — the "native residue" from the issue was this stranding plus its allocator fragmentation, not a separate PyO3 leak. Control (Transfer-resume) loop unchanged at 0 growth. Regression coverage: arena unit tests for drop-reclaim, bounded slot_count over 100 abandon cycles, stale-drop-after-clear isolation, and no-reclaim-on-attach; a Python abort-loop test asserting bounded head fiber indices via k.to_dict() (red: max index 400 for 200 aborts; green: 2); a new G1 guard pinning SlotReclaimQueue to bare usize indices. Co-Authored-By: Claude Fable 5 --- packages/doeff-vm-core/src/arena.rs | 157 +++++++++++++++++++ packages/doeff-vm-core/src/continuation.rs | 46 +++++- packages/doeff-vm/tests/test_memory_stats.py | 41 +++++ tests/core/test_vm_fiber_ownership_g1.py | 16 ++ 4 files changed, 258 insertions(+), 2 deletions(-) diff --git a/packages/doeff-vm-core/src/arena.rs b/packages/doeff-vm-core/src/arena.rs index 29967d26..7d7fc006 100644 --- a/packages/doeff-vm-core/src/arena.rs +++ b/packages/doeff-vm-core/src/arena.rs @@ -1,13 +1,50 @@ //! 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>, +} + +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) { + 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 { + let mut queue = self.dropped_slot_indices.lock().unwrap(); + std::mem::take(&mut *queue) + } +} + pub struct FiberArena { fibers: Vec>, free_list: Vec, + /// 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, } impl FiberArena { @@ -15,10 +52,12 @@ impl FiberArena { 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" + ); + } } diff --git a/packages/doeff-vm-core/src/continuation.rs b/packages/doeff-vm-core/src/continuation.rs index 38149c3b..e1b297a5 100644 --- a/packages/doeff-vm-core/src/continuation.rs +++ b/packages/doeff-vm-core/src/continuation.rs @@ -20,9 +20,12 @@ //! 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; @@ -57,6 +60,14 @@ pub struct DetachedFiberChain { head: FiberId, last_fiber: FiberId, fibers: Vec, + /// 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>, } impl DetachedFiberChain { @@ -65,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) { + self.slot_reclaim_queue = Some(queue); + } + pub fn head(&self) -> FiberId { self.head } @@ -80,8 +99,10 @@ impl DetachedFiberChain { &self.fibers } - pub fn into_fibers(self) -> Vec { - self.fibers + pub fn into_fibers(mut self) -> Vec { + // 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) -> bool { @@ -269,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 // --------------------------------------------------------------------------- diff --git a/packages/doeff-vm/tests/test_memory_stats.py b/packages/doeff-vm/tests/test_memory_stats.py index f600b824..0ddd0935 100644 --- a/packages/doeff-vm/tests/test_memory_stats.py +++ b/packages/doeff-vm/tests/test_memory_stats.py @@ -64,3 +64,44 @@ def scenario(): ] assert doeff_vm.vm_live_counts() == before + + +def test_arena_slots_reclaimed_when_handler_abandons_continuation() -> None: + """#497: a handler that never resumes k (abort-style) drops the detached + chain; the chain's arena slots must return to the free list within the + run. Before reclamation every abort stranded ~2 vacant-reserved slots, + so head fiber indices grew ~2 per abort (max index ~2*N); with + reclamation the same few slots are reused and indices stay bounded. + """ + n_aborts = 200 + head_indices: list[int] = [] + + @do + def abort_handler(effect, k): + if isinstance(effect, SyntheticQuery): + head_indices.append(k.to_dict()["head"]) + return "aborted" + yield doeff_vm.Pass(effect, k) + + @do + def body(): + yield SyntheticQuery(key="x") + return "unreachable" + + @do + def scenario(): + result = None + for _ in range(n_aborts): + result = yield doeff_vm.WithHandler(abort_handler, body()) + return result + + before = doeff_vm.vm_live_counts() + + assert run(scenario()) == "aborted" + + assert doeff_vm.vm_live_counts() == before + assert len(head_indices) == n_aborts + assert max(head_indices) <= 8, ( + f"arena slots stranded: max head fiber index {max(head_indices)} " + f"after {n_aborts} aborted dispatches (expected bounded slot reuse)" + ) diff --git a/tests/core/test_vm_fiber_ownership_g1.py b/tests/core/test_vm_fiber_ownership_g1.py index 4ea837b4..7bb665c8 100644 --- a/tests/core/test_vm_fiber_ownership_g1.py +++ b/tests/core/test_vm_fiber_ownership_g1.py @@ -57,6 +57,22 @@ def test_g1_arena_exposes_explicit_detach_attach_operations() -> None: assert "DetachedFiberChain" in source +def test_g1_slot_reclaim_carries_indices_not_fibers() -> None: + """#497: a detached chain dropped without reattachment reports its + arena slot indices through `SlotReclaimQueue` so the arena can reuse + them. That channel is allocator bookkeeping, not fiber ownership: it + must carry bare slot indices (usize) only — never fibers, chains, or + continuations. Fibers still move into Continuation ownership at detach + and drop normally with the chain (ISSUE-VM-001 G1 / SPEC-VM-021).""" + arena_source = _runtime_source(ARENA_RS) + body = _struct_body(arena_source, "SlotReclaimQueue") + + assert "Mutex>" in body + assert "Fiber" not in body + assert "Chain" not in body + assert "Continuation" not in body + + def test_g1_dispatch_and_step_do_not_construct_queue_backed_continuations() -> None: source = "\n".join( [