diff --git a/reacton/_threads_test_utils.py b/reacton/_threads_test_utils.py
new file mode 100644
index 0000000..5aad2fb
--- /dev/null
+++ b/reacton/_threads_test_utils.py
@@ -0,0 +1,202 @@
+"""Shared helpers for the render hand-off tests.
+
+Every test forces its interleaving: a thread is paused at a fixed point with an Event (a
+component body, an effect, a widget constructor, one of reacton's own logger.info calls, or one
+settrace pause at a source line). Only the stress tests depend on the scheduler.
+Calls that can hang on a broken tree run on daemon Worker threads with a timeout, so a failing
+test fails instead of hanging the run.
+"""
+
+import inspect
+import logging
+import sys
+import threading
+import time
+import traceback
+
+import pytest
+
+from reacton import core
+
+TIMEOUT = 5.0
+SHORT = 1.0  # "did this call block?" (it returns in microseconds when it does not)
+
+
+class LogHook(logging.Handler):
+    """Call a callback at one of reacton's logger.info calls, on a given thread, once.
+
+    handle() is overridden: logging.Handler.handle() takes the handler lock, and a thread
+    paused in emit() would then block every other thread's log call.
+    """
+
+    def __init__(self):
+        super().__init__(level=logging.DEBUG)
+        self.rules = []
+
+    def on(self, thread, prefix, callback):
+        self.rules.append([thread, prefix, callback, False])
+
+    def handle(self, record):
+        self.emit(record)
+        return True
+
+    def emit(self, record):
+        msg = record.msg if isinstance(record.msg, str) else str(record.msg)
+        for rule in self.rules:
+            thread, prefix, callback, fired = rule
+            if not fired and threading.current_thread() is thread and msg.startswith(prefix):
+                rule[3] = True
+                callback()
+
+
+class EveryTimeHook(logging.Handler):
+    """Call callback() at every reacton log record that starts with one of prefixes, on any
+    thread except the one named `skip`."""
+
+    def __init__(self, prefixes, callback, skip):
+        super().__init__(level=logging.DEBUG)
+        self.prefixes, self.callback, self.skip = tuple(prefixes), callback, skip
+
+    def handle(self, record):
+        self.emit(record)
+        return True
+
+    def emit(self, record):
+        msg = record.msg if isinstance(record.msg, str) else str(record.msg)
+        if msg.startswith(self.prefixes) and threading.current_thread().name != self.skip:
+            self.callback()
+
+
+def install(handler):
+    logger = logging.getLogger("reacton")
+    saved = logger.level, logger.propagate
+    logger.addHandler(handler)
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+
+    def uninstall():
+        logger.removeHandler(handler)
+        logger.setLevel(saved[0])
+        logger.propagate = saved[1]
+
+    return uninstall
+
+
+@pytest.fixture
+def log_hook():
+    hook = LogHook()
+    uninstall = install(hook)
+    yield hook
+    uninstall()
+
+
+@pytest.fixture
+def helper_errors():
+    """Exceptions raised on other threads (a helper thread renders with handle_error=False)."""
+    errors = []
+    if hasattr(threading, "excepthook"):
+        old = threading.excepthook
+        threading.excepthook = lambda args: errors.append(args.exc_value)
+        try:
+            yield errors
+        finally:
+            threading.excepthook = old
+        return
+
+    old_run = threading.Thread.run
+
+    def run(thread):
+        try:
+            old_run(thread)
+        except BaseException as e:
+            errors.append(e)
+            raise
+
+    threading.Thread.run = run  # type: ignore[method-assign]
+    try:
+        yield errors
+    finally:
+        threading.Thread.run = old_run  # type: ignore[method-assign]
+
+
+class Worker(threading.Thread):
+    def __init__(self, target, name):
+        super().__init__(name=name, daemon=True)
+        self._target_f = target
+        self.error = None
+        self.error_tb = ""
+        self.result = None
+
+    def run(self):
+        try:
+            self.result = self._target_f()
+        except BaseException as e:  # noqa
+            self.error = e
+            self.error_tb = traceback.format_exc()
+
+
+def call_in_thread(f, name, timeout=SHORT):
+    """Run f on a new thread; return (finished within timeout, worker)."""
+    t = Worker(f, name)
+    t.start()
+    t.join(timeout)
+    return not t.is_alive(), t
+
+
+def stacks():
+    frames = sys._current_frames()
+    return "\n".join(f"--- {t.name}\n" + "".join(traceback.format_stack(frames[t.ident])[-6:]) for t in threading.enumerate() if t.ident in frames)
+
+
+def text(box):
+    return box.children[0].description
+
+
+def wait_until(predicate, timeout):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return predicate()
+
+
+class Injected(Exception):
+    pass
+
+
+def render_loop_function():
+    """The function that holds the render loop (RC.render, or _render_locked when present)."""
+    RC = core._RenderContext
+    return getattr(RC, "_render_locked", None) or RC.render
+
+
+def source_line(fn, needle, after=None):
+    lines, first = inspect.getsourcelines(fn)
+    start = 0 if after is None else next(i for i, line in enumerate(lines) if after in line)
+    return first + next(i for i, line in enumerate(lines) if i >= start and needle in line)
+
+
+def pause_at_line(fn, lineno, on_pause, after_line=False):
+    """A sys.settrace function that calls on_pause() once, when fn executes line `lineno`
+    (after_line=True: at the first line event after `lineno` ran)."""
+    code = fn.__code__
+    state = {"seen": False, "fired": False}
+
+    def local_trace(frame, event, arg):
+        if event == "line" and not state["fired"]:
+            if after_line:
+                if state["seen"] and frame.f_lineno != lineno:
+                    state["fired"] = True
+                    on_pause()
+                elif frame.f_lineno == lineno:
+                    state["seen"] = True
+            elif frame.f_lineno == lineno:
+                state["fired"] = True
+                on_pause()
+        return local_trace
+
+    def global_trace(frame, event, arg):
+        return local_trace if frame.f_code is code else None
+
+    return global_trace
diff --git a/reacton/core.py b/reacton/core.py
index d4952ce..8b17c40 100644
--- a/reacton/core.py
+++ b/reacton/core.py
@@ -88,34 +88,47 @@ def get():
 # via the widget constructed hook: diffing the global widgets dict per widget
 # creation is O(total widgets), and solara replaces that dict with a context
 # aware mapping we should not depend on.
-_construction_recording: Optional[List["widgets.Widget"]] = None
+# The hook runs on the thread that constructs the widget, so the recording is per
+# thread (local.construction_recording): no lock is held while a widget is
+# constructed. Construction opens a comm, and in a server that send can wait for a
+# slow client; a process wide lock there stalled every other user (solara P3).
 _chained_construction_callback: Optional[Callable] = None
 
 
 def _record_constructed_widget(widget: "widgets.Widget"):
-    if _construction_recording is not None:
-        _construction_recording.append(widget)
+    recording = getattr(local, "construction_recording", None)
+    if recording is not None:
+        recording.append(widget)
     if _chained_construction_callback is not None:
         _chained_construction_callback(widget)
 
 
-def _start_recording_constructed(recording: List["widgets.Widget"]):
-    global _construction_recording, _chained_construction_callback
-    current = getattr(widgets.Widget, "_widget_construction_callback", None)
-    if current is not _record_constructed_widget:
-        # first time, or someone else registered a callback after us: chain it
-        _chained_construction_callback = current
-        widgets.Widget.on_widget_constructed(_record_constructed_widget)
-    _construction_recording = recording
+def _start_recording_constructed(recording: List["widgets.Widget"]) -> Optional[List["widgets.Widget"]]:
+    """Record the widgets constructed on this thread. Returns the recording it replaces
+    (a widget whose constructor renders reacton nests one _create_widget in another)."""
+    global _chained_construction_callback
+    with Element.create_lock:  # a leaf lock now: only guards installing the hook
+        current = getattr(widgets.Widget, "_widget_construction_callback", None)
+        if current is not _record_constructed_widget:
+            # first time, or someone else registered a callback after us: chain it
+            _chained_construction_callback = current
+            widgets.Widget.on_widget_constructed(_record_constructed_widget)
+    previous = getattr(local, "construction_recording", None)
+    local.construction_recording = recording
+    return previous
 
 
-def _stop_recording_constructed():
-    global _construction_recording
-    _construction_recording = None
+def _stop_recording_constructed(previous: Optional[List["widgets.Widget"]] = None):
+    local.construction_recording = previous
 
 
 _last_rc = None  # used for testing
 local = threading.local()
+# One call (a state change, a batch exit, render()) takes the render lock at most this many
+# times for render requests of other threads; then a helper thread renders the rest (see
+# _drain_render_requests).
+_MAX_FOREIGN_RENDERS = 50
+_NOT_TAKEN = object()  # _hold_and_render: another thread holds the render lock
 T = TypeVar("T")
 U = TypeVar("U")
 W = TypeVar("W")  # used for widgets
@@ -334,6 +347,7 @@ class Element(Generic[W]):
     # to make every unique on_value callback to a unique wrapper
     # so that we can remove the listeners
     _callback_wrappers: Dict[Tuple[str, str, Callable], Callable] = {}
+    # a leaf lock: only guards installing the widget construction hook (not held while a widget is constructed)
     create_lock: ContextManager = threading.Lock()
     _shared = False
 
@@ -499,32 +513,32 @@ def _create_widget(self, kwargs):
         # we can't use our own kwarg, since that contains elements, not widgets
         kwargs, listeners = self._split_kwargs(kwargs)
         assert isinstance(self.component, ComponentWidget)
-        # The recording is global state, so we need a lock.
-        with self.create_lock:
-            rc = get_render_context(required=True)
-            recorded: List[widgets.Widget] = []
-            _start_recording_constructed(recorded)
+        # No lock around the construction: the recording is per thread (see
+        # _record_constructed_widget), and the construction may block in a send.
+        rc = get_render_context(required=True)
+        recorded: List[widgets.Widget] = []
+        previous_recording = _start_recording_constructed(recorded)
+        try:
             try:
-                try:
-                    widget = self.component.widget(**kwargs)
-                    hold_trait_notifications = widget.hold_trait_notifications
-
-                    @contextlib.contextmanager
-                    def hold_trait_notifications_extra(*args, **kwargs):
-                        with rc, hold_trait_notifications(*args, **kwargs):
-                            yield
-
-                    widget.hold_trait_notifications = hold_trait_notifications_extra
-
-                    if self._meta:
-                        widget._react_meta = dict(self._meta)
-                except Exception as e:
-                    raise RuntimeError(f"Could not create widget {self.component.widget} with {kwargs}") from e
-                for name, callback in listeners.items():
-                    if callback is not None:
-                        self._add_widget_event_listener(widget, name, callback)
-            finally:
-                _stop_recording_constructed()
+                widget = self.component.widget(**kwargs)
+                hold_trait_notifications = widget.hold_trait_notifications
+
+                @contextlib.contextmanager
+                def hold_trait_notifications_extra(*args, **kwargs):
+                    with rc, hold_trait_notifications(*args, **kwargs):
+                        yield
+
+                widget.hold_trait_notifications = hold_trait_notifications_extra
+
+                if self._meta:
+                    widget._react_meta = dict(self._meta)
+            except Exception as e:
+                raise RuntimeError(f"Could not create widget {self.component.widget} with {kwargs}") from e
+            for name, callback in listeners.items():
+                if callback is not None:
+                    self._add_widget_event_listener(widget, name, callback)
+        finally:
+            _stop_recording_constructed(previous_recording)
         widgets_dict = _get_widgets_dict()
         orphans = {w.model_id for w in recorded if w is not widget and w.comm is not None and w.model_id in widgets_dict}
         return widget, orphans
@@ -1344,7 +1358,16 @@ def __init__(self, element: Element, container: widgets.Widget = None, children_
         self._lock_thread = cast(Optional[threading.Thread], None)
         self.last_root_widget: widgets.Widget = None
         self._is_rendering = False
+        # a re-render asked for by the thread that holds thread_lock (its own setters, effects,
+        # exception paths); only these count towards the "too many renders" limit
         self._rerender_needed = False
+        # a render asked for by a thread that does not hold thread_lock. It never waits for the
+        # lock: the holder renders it, because every holder checks this again after it released
+        # the lock (see _request_render and _drain_render_requests)
+        self._render_requested = False
+        # the root element to render: update() writes it on any thread, then requests a render;
+        # every pass reads it into self.element after it cleared the marks
+        self._element_requested: Optional[Element] = element
         # the reasons are only read for the "too many renders" error message, and a reason
         # holds the previous and next state value: keeping all of them kept every old state
         # value alive until close(). REACTON_RERENDER_REASONS keeps more, for debugging.
@@ -1352,6 +1375,8 @@ def __init__(self, element: Element, container: widgets.Widget = None, children_
         self._rerender_needed_reasons: Deque[RerenderReason] = collections.deque(maxlen=max_reasons)
         self.thread_lock = threading.Lock()
         self._closing = False
+        self._close_claim = threading.Lock()
+        self._teardown_done = False
         self.tracebacks: List[TracebackType] = []
         self.handle_error = handle_error
         self.reconsolidating = False
@@ -1388,9 +1413,8 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_value, traceback):
         counter = self._batch_counter.decrement()
         if counter == 0:
-            logger.info("finishing batch render (%s)", "needs rerender" if self._rerender_needed else "no rerender needed")
-            if self._rerender_needed:
-                self._possible_rerender()
+            logger.info("finishing batch render (%s)", "needs rerender" if self._render_requested else "no rerender needed")
+            self._drain_render_requests()  # requests made during the batch were only marked
 
     def find(self, cls: Type[W] = ipywidgets.Widget, **matches):
         from .find import finder
@@ -1400,50 +1424,66 @@ def find(self, cls: Type[W] = ipywidgets.Widget, **matches):
     _find = find  # for backward compatibility
 
     def close(self):
-        with self.thread_lock:
+        if self._teardown_done:
+            return
+        if self._lock_thread is threading.current_thread():
+            # thread_lock is not reentrant: waiting for it here would wait for ourselves
+            raise RuntimeError("close() called from a render of the same render context (avoided deadlock)")
+        if not self._close_claim.acquire(blocking=False):
+            return  # another close() is running, possibly from one of its cleanups
+        try:
+            # Before we wait for the lock: a render on another thread stops re-rendering and
+            # requests are dropped from now on, so we wait for at most the rest of the pass in flight.
             self._closing = True
-            # snapshot the component contexts before _remove_element detaches them from
-            # their parents: detached contexts would escape the teardown below while the
-            # setter/handler closures in their state still reference them and us
-            all_contexts: List[ComponentContext] = []
-
-            def collect(context: ComponentContext):
-                all_contexts.append(context)
-                for child in list(context.children.values()) + list(context.children_next.values()):
-                    collect(child)
-
-            collect(self.context_root)
-            logger.info("Removing elements...")
-            self._remove_element(self.element, default_key="/", parent_key=ROOT_KEY)
-            logger.info("Removing elements done.")
-            assert self.context is self.context_root
-            # everything below used to run outside the lock: a render() that was
-            # blocked on the lock could then interleave with this teardown and
-            # find self.context None mid-render (AssertionError in
-            # Element.__exit__, seen in production behind a prompt kernel close)
-            if self.container:
-                self.container.close()
-                if isinstance(self.container, widgets.DOMWidget) and self.container.layout is not None:
-                    self.container.layout.close()
-            if self._shared_elements:
-                raise RuntimeError(f"Element not cleaned up: {self._shared_elements}")
-            if self._orphans:
-                orphan_widgets = set([_get_widgets_dict()[k] for k in self._orphans])
-                raise RuntimeError(f"Orphan widgets not cleaned up for widgets: {orphan_widgets}")
-            exceptions = [*self.context.exceptions_children, *self.context_root.exceptions_self]
-            # break the reference cycles through the tree (see _teardown_component_context);
-            # _closing stays True, making stray setters and event handlers no-ops
-            for context in all_contexts:
-                _teardown_component_context(context)
-            self.context = None
-            self.context_root = None  # type: ignore
-            # the root element, container and root widget keep the widget tree alive, and
-            # widgets reference their elements, whose kwargs hold user callbacks - which
-            # capture use_state setters and therefore this render context
-            self.element = None  # type: ignore
-            self.container = None
-            self.last_root_widget = None
-            self._old_element_ids.clear()
+            with self.thread_lock:
+                if self._teardown_done:
+                    return  # two close() calls raced past the check above
+                # snapshot the component contexts before _remove_element detaches them from
+                # their parents: detached contexts would escape the teardown below while the
+                # setter/handler closures in their state still reference them and us
+                all_contexts: List[ComponentContext] = []
+
+                def collect(context: ComponentContext):
+                    all_contexts.append(context)
+                    for child in list(context.children.values()) + list(context.children_next.values()):
+                        collect(child)
+
+                collect(self.context_root)
+                logger.info("Removing elements...")
+                self._remove_element(self.element, default_key="/", parent_key=ROOT_KEY)
+                logger.info("Removing elements done.")
+                assert self.context is self.context_root
+                # everything below used to run outside the lock: a render() that was
+                # blocked on the lock could then interleave with this teardown and
+                # find self.context None mid-render (AssertionError in
+                # Element.__exit__, seen in production behind a prompt kernel close)
+                if self.container:
+                    self.container.close()
+                    if isinstance(self.container, widgets.DOMWidget) and self.container.layout is not None:
+                        self.container.layout.close()
+                if self._shared_elements:
+                    raise RuntimeError(f"Element not cleaned up: {self._shared_elements}")
+                if self._orphans:
+                    orphan_widgets = set([_get_widgets_dict()[k] for k in self._orphans])
+                    raise RuntimeError(f"Orphan widgets not cleaned up for widgets: {orphan_widgets}")
+                exceptions = [*self.context.exceptions_children, *self.context_root.exceptions_self]
+                # break the reference cycles through the tree (see _teardown_component_context);
+                # _closing stays True, making stray setters and event handlers no-ops
+                for context in all_contexts:
+                    _teardown_component_context(context)
+                self.context = None
+                self.context_root = None  # type: ignore
+                # the root element, container and root widget keep the widget tree alive, and
+                # widgets reference their elements, whose kwargs hold user callbacks - which
+                # capture use_state setters and therefore this render context
+                self.element = None  # type: ignore
+                self.container = None
+                self.last_root_widget = None
+                self._old_element_ids.clear()
+                self._element_requested = None
+                self._teardown_done = True
+        finally:
+            self._close_claim.release()
         if exceptions:
             raise exceptions[0]
 
@@ -1579,8 +1619,7 @@ def set_(value):
                         )
                     else:
                         self._rerender_needed_reasons.append(RerenderReason(reason=f"state changed with key {key}", prev_value=prev_value, next_value=value))
-                    self._rerender_needed = True
-                self._possible_rerender()
+                self._request_render()
 
         return set_
 
@@ -1590,8 +1629,9 @@ def force_update(self):
             return
         # a forced update re-walks the whole tree, no subtree skipping
         self._walk_all = True
-        if not self._is_rendering:
-            self.render(self.element, self.container)
+        if self._is_rendering and self._lock_thread is threading.current_thread():
+            return  # from our own render pass: unchanged, that pass walks the tree
+        self._request_render()  # before: dropped while another thread rendered
 
     def use_effect(self, effect: EffectCallable, dependencies=None):
         assert self.context is not None
@@ -1613,19 +1653,65 @@ def use_effect(self, effect: EffectCallable, dependencies=None):
             self.context.effect_index += 1
 
     def update(self, element: Element):
+        if self._closing:
+            return
         self._walk_all = True
-        if self._is_rendering:
-            self.element = element
-            self._rerender_needed_reasons.append(RerenderReason(reason="root element changed"))
-            self._rerender_needed = True
-        else:
-            self.render(element, self.container)
+        # not self.element: the thread that holds the lock may be in the middle of a pass
+        self._element_requested = element
+        self._rerender_needed_reasons.append(RerenderReason(reason="root element changed"))
+        self._request_render()
 
-    def _possible_rerender(self):
-        if not self._is_rendering and self._batch_counter.current() == 0:
-            self.render(self.element, self.container)
-        else:
-            logger.info("No render phase triggered, already rendering")
+    def _request_render(self):
+        """Ask for a render. Never waits for thread_lock, and the request is never lost."""
+        if self._closing:
+            return
+        if self._lock_thread is threading.current_thread():
+            if self._is_rendering:
+                self._rerender_needed = True  # our own render loop picks it up (and counts it)
+            else:
+                self._render_requested = True  # our own drain, after the release, picks it up
+            return
+        self._render_requested = True  # must come before the try-acquire in the drain
+        self._drain_render_requests()
+
+    def _drain_render_requests(self, budget: Optional[int] = _MAX_FOREIGN_RENDERS):
+        """Render while a render is requested, unless another thread holds thread_lock.
+
+        Returning when the lock is taken is safe: every thread that held the lock calls this
+        after it released the lock, and a requester sets _render_requested before it tries the
+        lock, so one of the two sees the other. Inside a batch (or while a thread waits in
+        render(), which counts as a batch) we only mark: that thread renders when it is done.
+
+        budget: how many more times this thread may take the lock for requests before it hands
+        the rest to a helper thread, so that its own caller (say, an event handler) is not held
+        up by a thread that keeps changing state. None: no limit. One hold renders the requests
+        made before it started, and its own re-renders; a request made during the hold ends it
+        (see _more_passes), and this loop takes the lock again for it.
+        """
+        while self._render_requested and not self._closing and self._batch_counter.current() == 0:
+            try:
+                if budget is not None and budget <= 0:
+                    if self._start_render_helper():
+                        return
+                    budget = None  # no threads (pyodide) or interpreter shutdown: render here
+                if self._hold_and_render(None, self.container, wait=False) is _NOT_TAKEN:
+                    logger.info("No render phase triggered, another thread holds the render lock and will pick it up")
+                    return
+                if budget is not None:
+                    budget -= 1
+            except BaseException:
+                if self._render_requested and not self._closing:
+                    self._start_render_helper()
+                raise
+
+    def _start_render_helper(self) -> bool:
+        logger.info("a helper thread renders the pending render requests")
+        try:
+            # no budget: a helper has no caller to hold up
+            threading.Thread(target=self._drain_render_requests, kwargs={"budget": None}, name="reacton-render", daemon=True).start()
+        except RuntimeError:
+            return False
+        return True
 
     def _discard_aborted_pass(self):
         """Forget everything a render pass staged when it raised before reconciliation.
@@ -1657,189 +1743,243 @@ def _discard_aborted_pass(self):
         self.context = self.context_root
 
     def render(self, element: Element, container: widgets.Widget = None):
-        # render + consolidate
-        widget = None
+        """Render element now, and return the root widget.
+
+        The only render entry that waits for thread_lock: it waits for a render (or close) in
+        progress on another thread. State changes, update() and force_update() go through
+        _request_render instead, and never wait.
+        """
         if container is None:
             container = self.container
-        was_locked = False
-        if self.thread_lock.locked():
-            if self._lock_thread == threading.current_thread():
-                raise RuntimeError("Recursive render detected (avoided deadlock), current thread: %r" % threading.current_thread())
-            logger.info(
-                "Render phase still in progress, waiting for mutex to release (locked obtained by %r, we are in thread %r)",
-                self._lock_thread,
-                threading.current_thread(),
-            )
-            was_locked = True
-        with self.thread_lock:
-            self._lock_thread = threading.current_thread()
-            if was_locked:
-                logger.info("Mutex released, continuing render phase")
-            if self._closing or self.context is None:
-                # close() won the race for the lock (a disconnect can close the
-                # kernel while an update was waiting to render): the tree is
-                # torn down, there is nothing to render into anymore
-                logger.info("Render requested on a closing/closed render context, ignoring")
-                return container
-            prev_rc = getattr(local, "rc", None)
-            # an exception that escapes while this is True aborted a render pass (see the except below)
-            in_render_phase = True
-            try:
-                local.rc = self
-                self.element = element
-                del element
-                main_render_phase = not self._is_rendering
-                render_count = self.render_count  # make a copy
-                self._rerender_needed = False
-                logger.info("Render phase: %r %r of %r", self.render_count, "main" if main_render_phase else "(nested)", self.element)
-                self.render_count += 1
-                self._is_rendering = True
-                # if we got called recursively, self.context is not the root context
-                context_prev = self.context
-                self.context = self.context_root
-                self.context.exception_handler = False
-                self.context.exceptions_children = []
-                self.context.exceptions_self = []
-                self.context.root_element_next = self.element
-                assert self.context is not None
+        if self._closing:
+            logger.info("Render requested on a closing/closed render context, ignoring")
+            return container
+        if self._lock_thread is threading.current_thread():
+            raise RuntimeError("Recursive render detected (avoided deadlock), current thread: %r" % threading.current_thread())
+        try:
+            widget = self._hold_and_render(element, container, wait=True)
+            self._drain_render_requests()  # requests that came after our last check
+        except BaseException:
+            if self._render_requested and not self._closing:
+                self._start_render_helper()
+            raise
+        return widget
 
+    def _hold_and_render(self, element: Optional[Element], container: Optional[widgets.Widget], wait: bool):
+        """Take thread_lock, render until nothing of our own is pending, and release the lock.
+
+        element None: render the root element (the one from update(), if any). Returns
+        _NOT_TAKEN when wait is False and another thread holds the lock, else the root widget.
+        The caller must drain after this returns (a request may have come during the hold).
+        """
+        acquired = False
+        try:
+            # the acquire is inside the try: an exception right after it (a KeyboardInterrupt, or
+            # an exception raised by a trace function) must not leak the lock
+            acquired = self.thread_lock.acquire(blocking=False)
+            if not acquired and wait:
+                logger.info(
+                    "Render phase still in progress, waiting for mutex to release (locked obtained by %r, we are in thread %r)",
+                    self._lock_thread,
+                    threading.current_thread(),
+                )
+                # A batch while we wait: other threads leave their render requests to us.
+                counter = None
                 try:
-                    self._shared_elements_next = set()
-                    self._render(self.element, "/", parent_key=ROOT_KEY)
-                    self.first_render = False
-                    self._walk_all = False
-                except BaseException:
-                    self._is_rendering = False
-                    raise
+                    counter = self._batch_counter.increment()
+                    if counter == 1:
+                        logger.info("entering batch render")
+                    acquired = self.thread_lock.acquire()
+                finally:
+                    if counter is not None:
+                        counter = self._batch_counter.decrement()
+                        if counter == 0:
+                            logger.info("finishing batch render (%s)", "needs rerender" if self._render_requested else "no rerender needed")
+                            self._drain_render_requests()
+            if not acquired:
+                return _NOT_TAKEN
+            self._lock_thread = threading.current_thread()
+            return self._render_locked(element, container)
+        finally:
+            if acquired:
+                self._lock_thread = None  # before the release, so the recursion guards see the holder only
+                self.thread_lock.release()
+
+    def _render_locked(self, element: Optional[Element], container: Optional[widgets.Widget]):
+        # render + consolidate, with thread_lock held by this thread (see _hold_and_render)
+        widget = None
+        if self._closing or self.context is None:
+            # close() won the race for the lock (a disconnect can close the
+            # kernel while an update was waiting to render): the tree is
+            # torn down, there is nothing to render into anymore
+            logger.info("Render requested on a closing/closed render context, ignoring")
+            return container
+        prev_rc = getattr(local, "rc", None)
+        # an exception that escapes while this is True aborted a render pass (see the except below)
+        in_render_phase = True
+        try:
+            local.rc = self
+            # consume the requests before this pass reads any state; the root element is state too.
+            # Only here, at the start of a hold: a request that comes later ends the hold (see
+            # _more_passes), and the drain after the release takes the lock again for it.
+            # (Clearing it anywhere else would hide that request from _more_passes, and the
+            # limit would count it as our own render loop.)
+            self._render_requested = False
+            self._rerender_needed = False
+            if element is not None:
+                self._element_requested = element  # an update() from before this render() is superseded
+            self.element = self._element_requested  # type: ignore  # None only after close(), checked above
+            del element
+            main_render_phase = not self._is_rendering
+            render_count = self.render_count  # make a copy
+            logger.info("Render phase: %r %r of %r", self.render_count, "main" if main_render_phase else "(nested)", self.element)
+            self.render_count += 1
+            self._is_rendering = True
+            # if we got called recursively, self.context is not the root context
+            context_prev = self.context
+            self.context = self.context_root
+            self.context.exception_handler = False
+            self.context.exceptions_children = []
+            self.context.exceptions_self = []
+            self.context.root_element_next = self.element
+            assert self.context is not None
 
-                if main_render_phase:
-                    stable = False
-                    render_counts = 0
-                    while not stable and not self.context_root.exceptions_children:
-                        # we started the rendering loop (main_render_phase is True), so we keep going
-                        # but if an exception bubbled up, we should stop
-                        while self._rerender_needed and not self.context_root.exceptions_children:
-                            if render_counts > 50:
-
-                                def format(reason: RerenderReason):
-                                    f = f"Reason: {reason.reason}\nValue changed from {reason.prev_value} to {reason.next_value}\n"
-                                    if reason.created_stack:
-                                        f += f"Created at: {''.join(reason.created_stack)}\n"
-                                    if reason.trigger_stack:
-                                        f += f"Triggered at: {''.join(reason.trigger_stack)}\n"
-                                    return f
-
-                                msg = f"Too many renders triggered, your render loop does not stop\nLast reason: {format(self._rerender_needed_reasons[-1])}\n"
-                                if len(self._rerender_needed_reasons) >= 2:
-                                    previous = reversed(list(self._rerender_needed_reasons)[:-1])
-                                    msg += f"Previous reasons: {''.join(format(reason) for reason in previous)}\n"
-                                raise RuntimeError(msg)
-                            logger.info("Entering nested render phase: %r", self._rerender_needed_reasons[-1])
-                            self._rerender_needed = False
-                            self._shared_elements_next = set()
-                            self.context.exception_handler = False
-                            self.context.exceptions_children = []
-                            self.context.exceptions_self = []
-
-                            self._render(self.element, "/", parent_key=ROOT_KEY)
-                            self._walk_all = False
-                            logger.info("Render done: %r %r", self._rerender_needed, self._rerender_needed_reasons[-1])
-                            assert self.context is self.context_root
-                            render_counts += 1
-                        logger.debug("Render phase resulted in (next) elements:")
-                        for el in self._shared_elements_next:
-                            logger.debug("\t%r %x", el, id(el))
-
-                        logger.debug("Current elements:")
-                        for el in self._shared_elements:
-                            logger.debug("\t %r %x", el, id(el))
-                        if self.context_root.exceptions_children:
-                            # an exception bubbled up render
-                            break
-
-                        logger.info("Render reconsolidate...")
+            self._shared_elements_next = set()
+            self._render(self.element, "/", parent_key=ROOT_KEY)
+            self.first_render = False
+            self._walk_all = False
+
+            if main_render_phase:
+                stable = False
+                render_counts = 0
+                # decided once and then acted on: other threads can make _more_passes() False
+                # (a request, close()), and a reconcile needs a render pass before it
+                more = self._more_passes(in_render_phase=True)
+                while not stable and not self.context_root.exceptions_children:
+                    # we started the rendering loop (main_render_phase is True), so we keep going
+                    # but if an exception bubbled up, we should stop
+                    while more and not self.context_root.exceptions_children:
+                        if render_counts > 50:
+
+                            def format(reason: RerenderReason):
+                                f = f"Reason: {reason.reason}\nValue changed from {reason.prev_value} to {reason.next_value}\n"
+                                if reason.created_stack:
+                                    f += f"Created at: {''.join(reason.created_stack)}\n"
+                                if reason.trigger_stack:
+                                    f += f"Triggered at: {''.join(reason.trigger_stack)}\n"
+                                return f
+
+                            msg = f"Too many renders triggered, your render loop does not stop\nLast reason: {format(self._rerender_needed_reasons[-1])}\n"
+                            if len(self._rerender_needed_reasons) >= 2:
+                                previous = reversed(list(self._rerender_needed_reasons)[:-1])
+                                msg += f"Previous reasons: {''.join(format(reason) for reason in previous)}\n"
+                            raise RuntimeError(msg)
+                        logger.info("Entering nested render phase: %r", self._rerender_needed_reasons[-1])
+                        self._rerender_needed = False
+                        if self._element_requested is not self.element:  # update() from our own effect
+                            self.element = self.context.root_element_next = self._element_requested  # type: ignore  # not closing
+                        self._shared_elements_next = set()
+                        self.context.exception_handler = False
+                        self.context.exceptions_children = []
+                        self.context.exceptions_self = []
+
+                        self._render(self.element, "/", parent_key=ROOT_KEY)
+                        self._walk_all = False
+                        logger.info("Render done: %r %r", self._rerender_needed, self._rerender_needed_reasons[-1])
+                        assert self.context is self.context_root
+                        render_counts += 1
+                        more = self._more_passes(in_render_phase=render_counts <= 50)
+                    logger.debug("Render phase resulted in (next) elements:")
+                    for el in self._shared_elements_next:
+                        logger.debug("\t%r %x", el, id(el))
+
+                    logger.debug("Current elements:")
+                    for el in self._shared_elements:
+                        logger.debug("\t %r %x", el, id(el))
+                    if self.context_root.exceptions_children:
+                        # an exception bubbled up render
+                        break
+
+                    logger.info("Render reconsolidate...")
+                    try:
                         self.reconsolidating = True
                         in_render_phase = False
-                        try:
-                            widget = self._reconsolidate(self.element, default_key="/", parent_key=ROOT_KEY)
-                        finally:
-                            self.reconsolidating = False
-                        in_render_phase = True
-                        logger.info("Render reconsolidate done")
-                        self.context.root_element = self.context.root_element_next
-                        self.context.root_element_next = None
-
-                        # remove stale elements of the root context itself
-                        # (child contexts are swept during their reconciliation)
-                        self._remove_stale_root_elements(ROOT_KEY)
-
-                        if self._shared_elements_next:
-                            raise RuntimeError(f"Element not reconsolidated: {self._shared_elements_next}")
-                        logger.debug("Reconsolidate phase resulted in elements:")
-                        for el in self._shared_elements:
-                            logger.debug("\t%r %x", el, id(el))
-                        # RESET
-                        assert self.context is self.context_root
-                        if self.element.is_shared:
-                            assert widget in self._shared_widgets.values()
-                        else:
-                            assert widget in self.context_root.widgets.values()
-                        if self.last_root_widget is None:
-                            self.last_root_widget = widget
+                        widget = self._reconsolidate(self.element, default_key="/", parent_key=ROOT_KEY)
+                    finally:
+                        self.reconsolidating = False
+                    in_render_phase = True
+                    logger.info("Render reconsolidate done")
+                    self.context.root_element = self.context.root_element_next
+                    self.context.root_element_next = None
+
+                    # remove stale elements of the root context itself
+                    # (child contexts are swept during their reconciliation)
+                    self._remove_stale_root_elements(ROOT_KEY)
+
+                    if self._shared_elements_next:
+                        raise RuntimeError(f"Element not reconsolidated: {self._shared_elements_next}")
+                    logger.debug("Reconsolidate phase resulted in elements:")
+                    for el in self._shared_elements:
+                        logger.debug("\t%r %x", el, id(el))
+                    # RESET
+                    assert self.context is self.context_root
+                    if self.element.is_shared:
+                        assert widget in self._shared_widgets.values()
+                    else:
+                        assert widget in self.context_root.widgets.values()
+                    if self.last_root_widget is None:
+                        self.last_root_widget = widget
+                    else:
+                        if container is None:
+                            if self.last_root_widget != widget:
+                                raise ValueError(
+                                    "You are not using a container, and the root component returned a new widget,"
+                                    "make sure your root component always returns the same component type"
+                                )
+                    if container:
+                        if widget is None:
+                            # Exception occurred, and we cannot render the widget
+                            container.children = []
                         else:
-                            if container is None:
-                                if self.last_root_widget != widget:
-                                    raise ValueError(
-                                        "You are not using a container, and the root component returned a new widget,"
-                                        "make sure your root component always returns the same component type"
-                                    )
-                        if container:
-                            if widget is None:
-                                # Exception occurred, and we cannot render the widget
-                                container.children = []
-                            else:
-                                container.children = [widget]
+                            container.children = [widget]
 
-                        if self.context_root.exceptions_children or self.context_root.exceptions_self:
-                            # an exception bubbled up during reconsolidate
-                            break
+                    if self.context_root.exceptions_children or self.context_root.exceptions_self:
+                        # an exception bubbled up during reconsolidate
+                        break
 
-                        if self._rerender_needed:
-                            logger.info("Need rerender after reconsolidation: %r", self._rerender_needed_reasons[-1])
-                            stable = False
-                        else:
-                            stable = True
+                    more = self._more_passes(in_render_phase=False)
+                    if more:
+                        logger.info("Need rerender after reconsolidation: %r", self._rerender_needed_reasons[-1])
+                        stable = False
+                    else:
+                        stable = True
 
-                    self._is_rendering = False
-                self.context = context_prev
-                logger.info("Done with render phase: %r", render_count)
-            except BaseException as e:
-                # Exceptions raised by components are collected in exceptions_self, so an
-                # exception here comes from the render machinery itself (duplicate key,
-                # hook misuse, ...) or is a cancellation, and aborted a pass halfway. Drop
-                # what that pass staged: otherwise the next render reconciles the last
-                # committed elements, but runs the effects the aborted pass chained, closed
-                # over elements that were never reconciled (get_widget then fails with
-                # "found in a previous render").
-                if in_render_phase:
-                    self._discard_aborted_pass()
-                if DEBUG:
-                    # construct a fake traceback (showing how the elements were constructed)
-                    if not self.tracebacks:
-                        raise
-                    e = _with_tracebacks(e, self.tracebacks)
-                    raise e
-                else:
+                self._is_rendering = False
+            self.context = context_prev
+            logger.info("Done with render phase: %r", render_count)
+        except BaseException as e:
+            # Exceptions raised by components are collected in exceptions_self, so an
+            # exception here comes from the render machinery itself (duplicate key,
+            # hook misuse, ...) or is a cancellation, and aborted a pass halfway. Drop
+            # what that pass staged: otherwise the next render reconciles the last
+            # committed elements, but runs the effects the aborted pass chained, closed
+            # over elements that were never reconciled (get_widget then fails with
+            # "found in a previous render").
+            if in_render_phase:
+                self._discard_aborted_pass()
+            if DEBUG:
+                # construct a fake traceback (showing how the elements were constructed)
+                if not self.tracebacks:
                     raise
+                e = _with_tracebacks(e, self.tracebacks)
+                raise e
+            else:
+                raise
 
-            finally:
-                local.rc = prev_rc  # type: ignore
-                self._is_rendering = False
-                # clear before the lock is released: a stale _lock_thread makes the
-                # recursion guard above fire for a thread that merely rendered last,
-                # while a *different* thread holds the lock (false "Recursive render")
-                self._lock_thread = None
-                assert self.context is self.context_root
+        finally:
+            local.rc = prev_rc  # type: ignore
+            self._is_rendering = False
+            assert self.context is self.context_root
 
         exceptions = [*self.context.exceptions_children, *self.context_root.exceptions_self]
         if exceptions:
@@ -1858,11 +1998,26 @@ def format(reason: RerenderReason):
                     value = html.escape(error)
                 from . import ipywidgets as w
 
-                return self.render(w.HTML(value="<pre>" + value + "</pre>", layout=w.Layout(overflow="auto")), self.container)
+                # in the same hold of the lock: no other render can come in between
+                return self._render_locked(w.HTML(value="<pre>" + value + "</pre>", layout=w.Layout(overflow="auto")), self.container)
             else:
                 raise exc
         return widget
 
+    def _more_passes(self, in_render_phase: bool) -> bool:
+        """Asked by the thread that renders, between passes: render another pass in this hold?
+
+        Only for our own re-renders (these count towards the limit). Inside the render phase we
+        go on, as before: a reconcile only commits a settled render. After a reconcile (and at
+        the limit) a request of another thread ends the hold instead: the drain after the
+        release renders it in a new hold, which respects batches, a waiting render() and the
+        budget (see _drain_render_requests). At the limit that also means: not a render loop of
+        ours, other threads keep changing state; commit what we have.
+        """
+        if self._closing or not self._rerender_needed:
+            return False
+        return in_render_phase or not self._render_requested
+
     def _render(self, element: Element, default_key: str, parent_key: str):
         if not isinstance(element, Element):
             raise TypeError(f"Expected element, not {element}")
diff --git a/reacton/threads_close_test.py b/reacton/threads_close_test.py
new file mode 100644
index 0000000..38ecb81
--- /dev/null
+++ b/reacton/threads_close_test.py
@@ -0,0 +1,374 @@
+"""close(): not starved, no render after it, no self-deadlock, requests during it dropped."""
+
+# ruff: noqa: F811  the fixture names shadow the pytest fixture parameters of the same names
+
+import sys
+import threading
+from typing import List
+
+import pytest
+
+import reacton
+import reacton.ipywidgets as w
+from reacton import core
+
+from ._threads_test_utils import TIMEOUT, Injected, Worker, call_in_thread, helper_errors, log_hook, stacks, wait_until  # noqa: F401
+
+
+def test_close_with_batch_in_cleanup_does_not_deadlock():
+    """A cleanup enters and leaves a batch (solara's store fire() does `with rc:`) while a render
+    request is still marked. Before: the batch exit called render() on close()'s own thread,
+    which waited for its own lock forever. Fails on 40a90a8."""
+    setters = {}
+
+    @reacton.component
+    def Test():
+        fail, setters["fail"] = reacton.use_state(False)
+        rc = core.get_render_context()
+
+        def effect():
+            def cleanup():
+                with rc:
+                    pass
+
+            return cleanup
+
+        reacton.use_effect(effect, [])
+        if fail:
+            raise ValueError("component fails")
+        return w.Button()
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+    with pytest.raises(ValueError):
+        setters["fail"](True)  # leaves a render marked (handle_error=False)
+    finished, _closer = call_in_thread(rc.close, "C", timeout=TIMEOUT)
+    assert finished, "close() deadlocked on its own render lock\n" + stacks()
+
+
+def test_close_is_not_starved_and_nothing_renders_after_close():
+    """K renders, T changes state during every pass (a fast progress reporter), C closes.
+    close() must not wait for T to stop, and no render pass may start after close() began
+    (the pass in flight may finish). Fails on 40a90a8."""
+    setters = {}
+    stop = threading.Event()
+    renders_after_close = []
+    closing = threading.Event()
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+    armed = threading.Event()
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        if closing.is_set():
+            renders_after_close.append(progress)
+        if armed.is_set() and threading.current_thread().name != "T" and not stop.is_set():
+            go.release()
+            done.acquire(timeout=TIMEOUT)
+        return w.Button(description=f"{trigger} {progress}")
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+    running = threading.Event()
+
+    def t_target():
+        i = 0
+        while not stop.is_set():
+            if not go.acquire(timeout=0.1):
+                continue
+            i += 1
+            setters["progress"](i)
+            if i == 3:
+                running.set()
+            done.release()
+
+    t = Worker(t_target, "T")
+    t.start()
+    armed.set()
+    k = Worker(lambda: setters["trigger"](1), "K")
+    k.start()
+    assert running.wait(TIMEOUT)
+
+    def close():
+        closing.set()
+        rc.close()
+
+    finished, c = call_in_thread(close, "C", timeout=TIMEOUT)
+    stop.set()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert finished, "close() waited for the progress reporter to stop\n" + stacks()
+    assert c.error is None, c.error
+    assert len(renders_after_close) <= 1, renders_after_close
+    assert k.error is None, k.error
+
+
+def test_close_from_own_render_raises_instead_of_hanging():
+    """Fails on 40a90a8 (hangs)."""
+    errors = []
+
+    @reacton.component
+    def Test():
+        rc = core.get_render_context()
+
+        def effect():
+            try:
+                rc.close()
+            except RuntimeError as e:
+                errors.append(e)
+
+        reacton.use_effect(effect, [])
+        return w.Button()
+
+    finished, _t = call_in_thread(lambda: reacton.render(Test(), handle_error=False), "K", timeout=TIMEOUT)
+    assert finished, "close() inside a render waited for its own lock\n" + stacks()
+    assert errors
+
+
+@pytest.mark.parametrize("what", ["close", "render"])
+def test_close_or_render_from_a_cleanup_during_close_does_not_hang(what):
+    """Fails on 40a90a8 (hangs)."""
+    rcs: dict = {}
+
+    @reacton.component
+    def Test():
+        def effect():
+            def cleanup():
+                rc = rcs["rc"]
+                if what == "close":
+                    rc.close()
+                else:
+                    rc.render(Test())
+
+            return cleanup
+
+        reacton.use_effect(effect, [])
+        return w.Button(description="x")
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+    rcs["rc"] = rc
+    finished, c = call_in_thread(rc.close, "C", timeout=TIMEOUT)
+    assert finished, f"close() hung when a cleanup called rc.{what}()\n" + stacks()
+    assert c.error is None, repr(c.error)
+
+
+def test_close_between_the_decision_to_render_again_and_the_pass(log_hook):
+    """K reconciled, an effect set state, K decided to render again; close() sets _closing
+    right then. K must not reconcile without a render pass (KeyError '/'), and close() must
+    finish. Fails on 40a90a8 (close() sets _closing only after it got the lock, and the
+    re-decision to render is not guarded)."""
+    setters = {}
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        derived, set_derived = reacton.use_state(0)
+        reacton.use_effect(lambda: set_derived(trigger), [trigger])
+        return w.Button(description=f"{trigger} {derived}")
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+    k_paused, resume = threading.Event(), threading.Event()
+
+    def k_target():
+        log_hook.on(threading.current_thread(), "Need rerender after reconsolidation", lambda: (k_paused.set(), resume.wait(TIMEOUT * 3)))  # type: ignore[func-returns-value]
+        setters["trigger"](1)
+
+    k = Worker(k_target, "K")
+    k.start()
+    assert k_paused.wait(TIMEOUT)
+    c = Worker(rc.close, "C")
+    c.start()
+    assert wait_until(lambda: rc._closing, TIMEOUT)
+    resume.set()
+    k.join(TIMEOUT)
+    c.join(TIMEOUT)
+    assert not k.is_alive() and not c.is_alive(), stacks()
+    assert k.error is None, k.error_tb or repr(k.error)
+    assert c.error is None, repr(c.error)
+
+
+def test_interrupted_close_can_be_retried():
+    """An interrupt after close() marks closing must not make every later close() return."""
+    setters = {}
+    k_paused, k_resume = threading.Event(), threading.Event()
+    cleanups = []
+
+    @reacton.component
+    def Test():
+        value, setters["value"] = reacton.use_state(0)
+
+        def effect():
+            return lambda: cleanups.append("cleanup")
+
+        reacton.use_effect(effect, [])
+        if value == 1 and threading.current_thread().name == "K":
+            k_paused.set()
+            k_resume.wait(TIMEOUT)
+        return w.Button(description=str(value))
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+    k = Worker(lambda: setters["value"](1), "K")
+    k.start()
+    assert k_paused.wait(TIMEOUT)
+    close_code = core._RenderContext.close.__code__
+    fired: List[str] = []
+
+    def trace(frame, event, arg):
+        if not fired and frame.f_code is close_code and rc._closing and rc.thread_lock.locked():
+            fired.append(event)
+            raise Injected()
+        return trace
+
+    def interrupted_close():
+        sys.settrace(trace)
+        try:
+            rc.close()
+        finally:
+            sys.settrace(None)
+
+    finished, c = call_in_thread(interrupted_close, "C", timeout=TIMEOUT)
+    assert finished, "the interrupted close() waited for the render lock\n" + stacks()
+    assert isinstance(c.error, Injected), repr(c.error)
+    assert fired, "the probe did not fire"
+    assert cleanups == []
+    k_resume.set()
+    k.join(TIMEOUT)
+    assert k.error is None, repr(k.error)
+    finished, c2 = call_in_thread(rc.close, "C2", timeout=TIMEOUT)
+    assert finished and c2.error is None, (c2.error, stacks())
+    assert cleanups == ["cleanup"]
+    finished, c3 = call_in_thread(rc.close, "C3", timeout=TIMEOUT)
+    assert finished and c3.error is None, (c3.error, stacks())
+    assert cleanups == ["cleanup"]
+
+
+def test_batched_set_from_another_thread_while_close_runs_cleanups(log_hook):
+    """The close ABBA. T is inside a setter (past the _closing check) when K starts close().
+    K's cleanup waits for T's batch to end (in solara: a store lock). Before: T's batch exit
+    waited for the render lock that close() holds: deadlock. Fails on 40a90a8."""
+    setters = {}
+    t_batch_done, t_paused, t_go = threading.Event(), threading.Event(), threading.Event()
+    cleanup_waited = []
+
+    @reacton.component
+    def Test():
+        value, setters["value"] = reacton.use_state(0)
+
+        def effect():
+            def cleanup():
+                t_go.set()
+                cleanup_waited.append(t_batch_done.wait(TIMEOUT))
+
+            return cleanup
+
+        reacton.use_effect(effect, [])
+        return w.Button(description=str(value))
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+
+    def t_target():
+        log_hook.on(threading.current_thread(), "Set state = ", lambda: (t_paused.set(), t_go.wait(TIMEOUT)))  # type: ignore[func-returns-value]
+        with rc:  # like a solara listener batch
+            setters["value"](1)
+        t_batch_done.set()
+
+    def k_target():
+        assert t_paused.wait(TIMEOUT)
+        rc.close()
+
+    t, k = Worker(t_target, "T"), Worker(k_target, "K")
+    t.start()
+    k.start()
+    k.join(TIMEOUT * 2)
+    t.join(TIMEOUT * 2)
+    assert not k.is_alive() and not t.is_alive(), "deadlock between close() and a batched set\n" + stacks()
+    assert k.error is None and t.error is None, (k.error, t.error)
+    assert cleanup_waited == [True], "the cleanup waited for T's batch, and T's batch exit waited for close()"
+
+
+def test_second_concurrent_close_does_not_deadlock_against_cleanup_lock():
+    ext = threading.RLock()
+    in_cleanup, c2_has_ext = threading.Event(), threading.Event()
+
+    @reacton.component
+    def Test():
+        def effect():
+            def cleanup():
+                in_cleanup.set()
+                assert c2_has_ext.wait(TIMEOUT)
+                with ext:
+                    pass
+
+            return cleanup
+
+        reacton.use_effect(effect, [])
+        return w.Button(description="x")
+
+    _box, rc = reacton.render(Test(), handle_error=False)
+
+    def c2_target():
+        assert in_cleanup.wait(TIMEOUT)
+        with ext:
+            c2_has_ext.set()
+            rc.close()
+
+    c1, c2 = Worker(rc.close, "C1"), Worker(c2_target, "C2")
+    c1.start()
+    c2.start()
+    c1.join(TIMEOUT)
+    c2.join(TIMEOUT)
+    assert not c1.is_alive() and not c2.is_alive(), "two close() calls deadlocked across a cleanup lock\n" + stacks()
+    assert c1.error is None and c2.error is None, (c1.error, c2.error)
+
+
+def test_requests_during_close_are_dropped_and_render_after_close_is_ignored(log_hook):
+    """Fails on 40a90a8 (T's requests wait for close())."""
+    setters = {}
+    rendered = []
+
+    @reacton.component
+    def Test():
+        value, setters["value"] = reacton.use_state(0)
+        rendered.append(value)
+        return w.Button(description=str(value))
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    k_paused, k_go = threading.Event(), threading.Event()
+
+    def k_target():
+        log_hook.on(threading.current_thread(), "Removing elements...", lambda: (k_paused.set(), k_go.wait(TIMEOUT * 3)))  # type: ignore[func-returns-value]
+        rc.close()
+
+    def t_target():
+        assert k_paused.wait(TIMEOUT)
+        setters["value"](1)  # must return at once and never render
+        rc.force_update()
+        rc.update(Test())
+
+    k = Worker(k_target, "K")
+    k.start()
+    finished, t = call_in_thread(t_target, "T", timeout=TIMEOUT)
+    k_go.set()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert finished, "a request on a closing context waited for close()\n" + stacks()
+    assert t.error is None and k.error is None, (t.error, k.error)
+    assert rendered == [0]
+    finished, r = call_in_thread(lambda: rc.render(Test(), box), "R", timeout=TIMEOUT)
+    assert finished and r.error is None, (r.error, stacks())
+    assert rendered == [0]
+
+
+def test_recursive_render_is_still_detected():
+    """A guard (passes on 40a90a8)."""
+
+    @reacton.component
+    def Page():
+        rc = core.get_render_context()
+        rc.render(rc.element, rc.container)
+        return w.Button(description="never")
+
+    finished, k = call_in_thread(lambda: reacton.render(Page(), handle_error=True), "K", timeout=TIMEOUT)
+    assert finished, stacks()
+    box, rc = k.result
+    assert "Recursive render detected" in box.children[0].value
+    rc.close()
diff --git a/reacton/threads_create_lock_test.py b/reacton/threads_create_lock_test.py
new file mode 100644
index 0000000..c0cffa0
--- /dev/null
+++ b/reacton/threads_create_lock_test.py
@@ -0,0 +1,110 @@
+"""Widget construction does not hold a shared lock: it does not block other render contexts or
+other threads, and a widget that renders in its constructor does not hang. Fail on 40a90a8."""
+
+import threading
+
+import ipywidgets
+import pytest
+
+import reacton
+import reacton.ipywidgets as w
+
+from ._threads_test_utils import TIMEOUT, Worker, call_in_thread, helper_errors, log_hook, stacks  # noqa: F401
+
+
+class BlockingButton(ipywidgets.Button):
+    """A widget whose construction blocks on thread "A", like comm_open to a slow client."""
+
+    entered = threading.Event()
+    release = threading.Event()
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        if threading.current_thread().name == "A":
+            BlockingButton.entered.set()
+            assert BlockingButton.release.wait(TIMEOUT)
+
+
+@pytest.fixture
+def blocking_button():
+    BlockingButton.entered.clear()
+    BlockingButton.release.clear()
+    yield BlockingButton
+    BlockingButton.release.set()
+
+
+def test_widget_construction_does_not_block_other_render_contexts(blocking_button):
+    """Before: one process-wide lock around every widget construction; a construction that
+    blocks (a send to a slow client) stops every other render context. Side-effect widgets
+    (Layout, Style) are still attributed to their own render context."""
+    BlockingButtonElement = reacton.core.ComponentWidget(blocking_button)
+
+    @reacton.component
+    def Slow():
+        return BlockingButtonElement(description="slow")
+
+    @reacton.component
+    def Fast():
+        return w.Button(description="fast")
+
+    a = Worker(lambda: reacton.render(Slow(), handle_error=False), "A")
+    a.start()
+    assert blocking_button.entered.wait(TIMEOUT)
+    finished, b = call_in_thread(lambda: reacton.render(Fast(), handle_error=False), "B")
+    blocking_button.release.set()
+    a.join(TIMEOUT)
+    b.join(TIMEOUT)
+    assert finished, "widget creation waited for another render context's widget construction\n" + stacks()
+    assert a.error is None and b.error is None, (a.error, b.error)
+    (box_a, rc_a), (box_b, rc_b) = a.result, b.result  # type: ignore[misc]
+    button_a, button_b = box_a.children[0], box_b.children[0]
+    assert rc_a._orphans[button_a.model_id] == {button_a.layout.model_id, button_a.style.model_id}
+    assert rc_b._orphans[button_b.model_id] == {button_b.layout.model_id, button_b.style.model_id}
+    rc_a.close()
+    rc_b.close()
+
+
+def test_widget_created_by_another_thread_is_not_an_orphan(blocking_button):
+    """Before: the construction recording was global, so a widget that another thread creates
+    while a render constructs a widget became an orphan of that widget, and was closed with it."""
+    BlockingButtonElement = reacton.core.ComponentWidget(blocking_button)
+
+    @reacton.component
+    def Slow():
+        return BlockingButtonElement(description="slow")
+
+    a = Worker(lambda: reacton.render(Slow(), handle_error=False), "A")
+    a.start()
+    assert blocking_button.entered.wait(TIMEOUT)
+    finished, u = call_in_thread(ipywidgets.IntSlider, "U")  # user code on another thread
+    blocking_button.release.set()
+    a.join(TIMEOUT)
+    assert finished, "creating a plain widget waited for another thread's widget construction\n" + stacks()
+    assert a.error is None, a.error
+    _box, rc = a.result  # type: ignore[misc]
+    rc.close()
+    unrelated = u.result
+    assert unrelated.comm is not None, "a widget of another thread was closed as an orphan"
+    unrelated.close()
+
+
+def test_widget_that_renders_in_its_constructor():
+    """Before: a widget class whose constructor renders reacton (ComponentFunction.widget_class)
+    used as an element took the non-reentrant create_lock twice on one thread: hang."""
+
+    @reacton.component
+    def Inner(label: str = "x"):
+        return w.Button(description=label)
+
+    InnerWidget = Inner.widget_class()  # type: ignore[attr-defined]
+
+    @reacton.component
+    def Outer():
+        return InnerWidget.element(label="nested")
+
+    finished, t = call_in_thread(lambda: reacton.render(Outer(), handle_error=False), "K", timeout=TIMEOUT)
+    assert finished, "nested widget construction deadlocked\n" + stacks()
+    assert t.error is None, t.error
+    box, rc = t.result
+    assert box.children[0].children[0].description == "nested"
+    rc.close()
diff --git a/reacton/threads_handoff_test.py b/reacton/threads_handoff_test.py
new file mode 100644
index 0000000..23cdc3e
--- /dev/null
+++ b/reacton/threads_handoff_test.py
@@ -0,0 +1,381 @@
+"""A render request (state change, update(), force_update(), batch exit) never waits for the
+render lock, and is never lost. Fails on 40a90a8 unless noted."""
+
+# ruff: noqa: F811  the fixture names shadow the pytest fixture parameters of the same names
+
+import sys
+import threading
+
+import reacton
+import reacton.ipywidgets as w
+from reacton import core
+
+from ._threads_test_utils import (  # noqa: F401
+    SHORT,
+    TIMEOUT,
+    Injected,
+    Worker,
+    call_in_thread,
+    helper_errors,
+    log_hook,
+    pause_at_line,
+    render_loop_function,
+    source_line,
+    stacks,
+    text,
+    wait_until,
+)
+
+
+def make_ab():
+    setters = {}
+
+    @reacton.component
+    def Test(label=""):
+        a, setters["a"] = reacton.use_state(0)
+        b, setters["b"] = reacton.use_state(0)
+        return w.Button(description=f"{label}{a} {b}")
+
+    return Test, setters
+
+
+def test_state_change_does_not_wait_for_a_render_on_another_thread_prologue(log_hook):
+    _test_state_change_does_not_wait(log_hook, "Render phase: ")
+
+
+def test_state_change_does_not_wait_for_a_render_on_another_thread_epilogue(log_hook):
+    _test_state_change_does_not_wait(log_hook, "Done with render phase")
+
+
+def _test_state_change_does_not_wait(log_hook, window):
+    """K holds the render lock outside its render loop (prologue or epilogue).
+    Before: T's setter waits for the render lock for the rest of K's render.
+    After: T's setter returns at once, and K renders T's change after it released the lock."""
+    Test, setters = make_ab()
+    box, rc = reacton.render(Test(), handle_error=False)
+    k_paused, t_done = threading.Event(), threading.Event()
+
+    def k_target():
+        log_hook.on(threading.current_thread(), window, lambda: (k_paused.set(), t_done.wait(TIMEOUT)))  # type: ignore[func-returns-value]
+        setters["a"](1)
+
+    k = Worker(k_target, "K")
+    k.start()
+    assert k_paused.wait(TIMEOUT)
+    finished, t = call_in_thread(lambda: setters["b"](1), "T")
+    t_done.set()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert finished, "the state change waited for another thread's render\n" + stacks()
+    assert k.error is None and t.error is None, (k.error, t.error)
+    assert text(box) == "1 1", "the state change from T was not rendered"
+    rc.close()
+
+
+def _pause_after_stable(on_pause):
+    fn = render_loop_function()
+    return pause_at_line(fn, source_line(fn, "self._is_rendering = False", after="stable = True"), on_pause)
+
+
+def test_state_change_after_the_last_check_is_not_lost():
+    """Lost wakeup. K has decided its render is stable and still holds the lock; T sets state
+    then. Before: T only marks, K releases, nobody renders T's change.
+    After: K checks again after it released the lock."""
+    Test, setters = make_ab()
+    box, rc = reacton.render(Test(), handle_error=False)
+    k_paused, t_done = threading.Event(), threading.Event()
+    tracer = _pause_after_stable(lambda: (k_paused.set(), t_done.wait(TIMEOUT)))  # type: ignore[func-returns-value]
+
+    def k_target():
+        sys.settrace(tracer)
+        try:
+            setters["a"](1)
+        finally:
+            sys.settrace(None)
+
+    k = Worker(k_target, "K")
+    k.start()
+    assert k_paused.wait(TIMEOUT)
+    finished, t = call_in_thread(lambda: setters["b"](1), "T")
+    t_done.set()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert finished and k.error is None and t.error is None, (k.error, t.error, stacks())
+    assert text(box) == "1 1", "the state change from T was lost"
+    rc.close()
+
+
+def test_state_change_after_the_last_check_of_an_explicit_render_is_not_lost():
+    """The same window, in an explicit render(): render() must check again after its release."""
+    Test, setters = make_ab()
+    box, rc = reacton.render(Test(), handle_error=False)
+    w_paused, t_done = threading.Event(), threading.Event()
+    tracer = _pause_after_stable(lambda: (w_paused.set(), t_done.wait(TIMEOUT)))  # type: ignore[func-returns-value]
+
+    def w_target():
+        sys.settrace(tracer)
+        try:
+            rc.render(Test(label="b"))
+        finally:
+            sys.settrace(None)
+
+    wt = Worker(w_target, "W")
+    wt.start()
+    assert w_paused.wait(TIMEOUT)
+    finished, t = call_in_thread(lambda: setters["b"](1), "T")
+    t_done.set()
+    wt.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert finished and wt.error is None and t.error is None, (wt.error, t.error, stacks())
+    assert wait_until(lambda: text(box) == "b0 1", 2), f"the state change from T was lost: {text(box)!r}"
+    rc.close()
+
+
+def test_lock_held_around_a_state_change_does_not_deadlock_with_an_effect(log_hook):
+    """A user lock (a solara store lock, init lock or context.lock behave the same).
+    T holds L while it sets state; K's render runs an effect that takes L.
+    Before: T waits for the render lock while it holds L, K waits for L: deadlock."""
+    user_lock = threading.RLock()
+    setters = {}
+
+    @reacton.component
+    def Test():
+        a, setters["a"] = reacton.use_state(0)
+        b, setters["b"] = reacton.use_state(0)
+
+        def effect():
+            if a == 1:
+                with user_lock:
+                    pass
+
+        reacton.use_effect(effect, [a])
+        return w.Button(description=f"{a} {b}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    k_paused, never = threading.Event(), threading.Event()
+
+    def k_target():
+        # pause in the render prologue: lock held, not yet rendering
+        log_hook.on(threading.current_thread(), "Render phase: ", lambda: (k_paused.set(), never.wait(SHORT)))  # type: ignore[func-returns-value]
+        setters["a"](1)
+
+    def t_target():
+        with user_lock:
+            setters["b"](1)
+
+    k = Worker(k_target, "K")
+    k.start()
+    assert k_paused.wait(TIMEOUT)
+    t = Worker(t_target, "T")
+    t.start()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert not k.is_alive() and not t.is_alive(), "deadlock\n" + stacks()
+    assert k.error is None and t.error is None, (k.error, t.error)
+    assert text(box) == "1 1"
+    rc.close()
+
+
+def _root_element_read():
+    """(function, line): the line where the first pass of a hold reads the root element."""
+    RC = core._RenderContext
+    if hasattr(RC, "_render_locked"):
+        fn = RC._render_locked
+        for needle in ("self.element = self._element_requested", "self.element = element"):
+            try:
+                return fn, source_line(fn, needle)
+            except StopIteration:
+                pass
+    return RC.render, source_line(RC.render, "self.element = element")
+
+
+def test_update_from_another_thread_at_the_start_of_a_pass_is_not_lost():
+    """K has just read the root element for its pass; T calls rc.update() with a new root.
+    T's element must be rendered (the pass start clears the request mark before it reads the
+    root element). Passes on 40a90a8 (T waits for the lock)."""
+    setters = {}
+
+    @reacton.component
+    def Test(label="old"):
+        a, setters["a"] = reacton.use_state(0)
+        return w.Button(description=f"{label} {a}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    fn, lineno = _root_element_read()
+    k_paused, t_done = threading.Event(), threading.Event()
+    tracer = pause_at_line(fn, lineno, lambda: (k_paused.set(), t_done.wait(SHORT)), after_line=True)  # type: ignore[func-returns-value]
+
+    def k_target():
+        sys.settrace(tracer)
+        try:
+            setters["a"](1)
+        finally:
+            sys.settrace(None)
+
+    k = Worker(k_target, "K")
+    k.start()
+    assert k_paused.wait(TIMEOUT)
+    finished, t = call_in_thread(lambda: rc.update(Test(label="new")), "T", timeout=SHORT)
+    t_done.set()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert k.error is None and t.error is None, (k.error, t.error)
+    assert wait_until(lambda: text(box) == "new 1", 2), f"the update() from T was lost: {text(box)!r} (T returned at once: {finished})"
+    rc.close()
+
+
+def test_update_from_an_effect_of_the_render_is_rendered():
+    """rc.update() from an effect of the render in progress (same thread): the next pass of
+    that render must use the new root element. Passes on 40a90a8."""
+    calls: list = []
+
+    @reacton.component
+    def Test(label="old"):
+        rc = core.get_render_context()
+
+        def effect():
+            if label == "old" and not calls:
+                calls.append(1)
+                rc.update(Test(label="new"))
+
+        reacton.use_effect(effect, [label])
+        return w.Button(description=label)
+
+    finished, k = call_in_thread(lambda: reacton.render(Test(), handle_error=False), "K", timeout=TIMEOUT)
+    assert finished and k.error is None, (k.error, stacks())
+    box, rc = k.result
+    assert text(box) == "new"
+    rc.close()
+
+
+def test_a_reconcile_commits_a_settled_render_only():
+    """A component derives state while it renders (`if b != a: set_b(a)`), so the pass after a
+    change of `a` is not settled. T asks for a render during that pass. The holder must still
+    finish its render phase before it reconciles: no effect may run on (a=1, b=0). Passes on
+    40a90a8."""
+    setters = {}
+    seen = []
+    k_in_pass, k_go = threading.Event(), threading.Event()
+
+    @reacton.component
+    def Test():
+        a, setters["a"] = reacton.use_state(0)
+        b, set_b = reacton.use_state(0)
+        x, setters["x"] = reacton.use_state(0)
+        if b != a:
+            if threading.current_thread().name == "K" and not k_in_pass.is_set():
+                k_in_pass.set()
+                k_go.wait(TIMEOUT)
+            set_b(a)
+        reacton.use_effect(lambda: seen.append((a, b)), [a, b])
+        return w.Button(description=f"{a} {b} {x}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    k = Worker(lambda: setters["a"](1), "K")
+    k.start()
+    assert k_in_pass.wait(TIMEOUT)
+    _finished, t = call_in_thread(lambda: setters["x"](1), "T", timeout=SHORT * 2)
+    k_go.set()
+    k.join(TIMEOUT)
+    t.join(TIMEOUT)
+    assert k.error is None and t.error is None, (k.error, t.error)
+    assert wait_until(lambda: text(box) == "1 1 1", 3), text(box)
+    assert (1, 0) not in seen, f"an effect ran on a render that was not settled: {seen}"
+    rc.close()
+
+
+def test_request_during_a_batch_on_another_thread_renders_at_batch_exit():
+    """While any thread is inside `with rc:`, a state change only marks, and the thread that
+    leaves the last batch renders. Passes on 40a90a8 (a guard)."""
+    Test, setters = make_ab()
+    box, rc = reacton.render(Test(), handle_error=False)
+    rc.__enter__()
+    finished, t = call_in_thread(lambda: setters["a"](1), "T")
+    assert finished and t.error is None
+    assert text(box) == "0 0", "rendered inside a batch"
+    rc.__exit__(None, None, None)
+    assert text(box) == "1 0"
+    rc.close()
+
+
+def test_exception_injected_right_after_the_lock_is_taken_does_not_leak_it():
+    """Models a KeyboardInterrupt, or solara's cancel_guard (a trace function that raises on a
+    traced line while the context is not rendering): it fires once, at the first traced event
+    in reacton/core.py after the thread took the render lock. The lock must not leak.
+    Passes on 40a90a8."""
+    Test, setters = make_ab()
+    _box, rc = reacton.render(Test(), handle_error=False)
+    corefile = core.__file__
+    fired: list = []
+
+    def trace(frame, event, arg):
+        if not fired and frame.f_code.co_filename == corefile and rc.thread_lock.locked() and not rc._is_rendering:
+            fired.append((frame.f_code.co_name, frame.f_lineno, event))
+            raise Injected()
+        return trace
+
+    def t_target():
+        sys.settrace(trace)
+        try:
+            setters["a"](1)
+        finally:
+            sys.settrace(None)
+
+    finished, t = call_in_thread(t_target, "T", timeout=TIMEOUT)
+    assert finished
+    assert isinstance(t.error, Injected), (t.error, fired)
+    assert not rc.thread_lock.locked(), f"the render lock leaked after an exception at {fired}"
+    finished, _c = call_in_thread(rc.close, "C", timeout=TIMEOUT)
+    assert finished, stacks()
+
+
+def test_concurrent_writers_never_hang_raise_or_lose_a_render():
+    """Stress, no pauses: K changes `trigger` (its effect writes `b` under a user lock), T writes
+    `a` under the same lock. No deadlock, no exception, the last state is shown (a helper may
+    render it a moment after the writers returned)."""
+    lock = threading.RLock()
+    setters = {}
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        a, setters["a"] = reacton.use_state(0)
+        b, set_b = reacton.use_state(0)
+
+        def effect():
+            with lock:  # like a solara store listener that holds the store lock
+                set_b(trigger)
+
+        reacton.use_effect(effect, [trigger])
+        return w.Button(description=f"{trigger} {a} {b}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    n = 2000
+    errors = []
+
+    def run(f):
+        try:
+            for i in range(1, n + 1):
+                f(i)
+        except BaseException as e:  # noqa
+            errors.append(e)
+
+    def write_a(i):
+        with lock:
+            setters["a"](i)
+
+    old = sys.getswitchinterval()
+    sys.setswitchinterval(1e-6)
+    try:
+        k = Worker(lambda: run(setters["trigger"]), "K")
+        t = Worker(lambda: run(write_a), "T")
+        k.start()
+        t.start()
+        k.join(60)
+        t.join(60)
+    finally:
+        sys.setswitchinterval(old)
+    assert not k.is_alive() and not t.is_alive(), "deadlock\n" + stacks()
+    assert not errors, errors
+    assert wait_until(lambda: text(box) == f"{n} {n} {n}", 10), text(box)
+    rc.close()
diff --git a/reacton/threads_limit_test.py b/reacton/threads_limit_test.py
new file mode 100644
index 0000000..a0107d1
--- /dev/null
+++ b/reacton/threads_limit_test.py
@@ -0,0 +1,233 @@
+"""The "too many renders" limit counts only a thread's own render loop; a caller is not held
+by other threads' updates (the budget and the helper); an explicit render() is not starved."""
+
+# ruff: noqa: F811  the fixture names shadow the pytest fixture parameters of the same names
+
+import threading
+import time
+
+import pytest
+
+import reacton
+import reacton.ipywidgets as w
+from reacton import core
+
+from ._threads_test_utils import TIMEOUT, Worker, call_in_thread, helper_errors, log_hook, stacks, text, wait_until  # noqa: F401
+
+
+@pytest.mark.parametrize("derive", ["none", "effect", "render"])
+def test_updates_from_another_thread_do_not_count_as_a_render_loop(derive):
+    """T changes state once during each of K's render passes, 60 times.
+    Before: K raises "Too many renders triggered" after about 50 passes.
+    derive="effect": an effect derives state from T's value, so every pass for T is followed by
+    an own pass. derive="render": the component derives it while it renders, so the render
+    phase does not settle while T keeps changing state; at the limit the holder commits what it
+    has instead of raising (T's requests are pending). Fails on 40a90a8."""
+    n = 60
+    setters = {}
+    remaining = [n]
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+    armed = threading.Event()
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        derived, set_derived = reacton.use_state(0)
+        if derive == "effect":
+            reacton.use_effect(lambda: set_derived(progress), [progress])
+        if derive == "render" and derived != progress:
+            set_derived(progress)
+        if armed.is_set() and threading.current_thread().name != "T" and remaining[0] > 0:  # K, or the helper
+            go.release()  # T reports progress once during this pass
+            assert done.acquire(timeout=TIMEOUT)
+        return w.Button(description=f"{trigger} {progress} {derived}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+
+    def t_target():
+        for i in range(n):
+            assert go.acquire(timeout=TIMEOUT)
+            setters["progress"](i + 1)
+            remaining[0] -= 1
+            done.release()
+
+    def k_target():
+        armed.set()
+        setters["trigger"](1)
+
+    t = Worker(t_target, "T")
+    t.start()
+    finished, k = call_in_thread(k_target, "K", timeout=TIMEOUT * 2)
+    t.join(TIMEOUT)
+    assert finished and not t.is_alive(), stacks()
+    assert k.error is None, repr(k.error)[:300]
+    assert t.error is None, repr(t.error)[:300]
+    expected = f"1 {n} {n if derive != 'none' else 0}"
+    assert wait_until(lambda: text(box) == expected, 5), text(box)
+    rc.close()
+
+
+@pytest.mark.parametrize("n", [80, 200])
+def test_foreign_updates_seen_by_an_effect_are_not_a_render_loop(n, helper_errors):
+    """T changes state inside every reconcile of the renderer (inside an effect), n times;
+    the effect derives own state from T's value. Not a render loop: no "Too many renders", also
+    not on the helper that renders after K's budget (n=200). Fails on 40a90a8."""
+    setters = {}
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+    armed = threading.Event()
+    remaining = [n]
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        derived, set_derived = reacton.use_state(0)
+
+        def effect():
+            set_derived(progress)
+            if armed.is_set() and threading.current_thread().name != "T" and remaining[0] > 0:
+                go.release()
+                done.acquire(timeout=TIMEOUT)
+
+        reacton.use_effect(effect, [progress, trigger])
+        return w.Button(description=f"{trigger} {progress} {derived}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+
+    def t_target():
+        for i in range(n):
+            assert go.acquire(timeout=TIMEOUT)
+            setters["progress"](i + 1)
+            remaining[0] -= 1
+            done.release()
+
+    def k_target():
+        armed.set()
+        setters["trigger"](1)
+
+    t = Worker(t_target, "T")
+    t.start()
+    finished, k = call_in_thread(k_target, "K", timeout=30)
+    t.join(30)
+    assert finished, stacks()
+    assert k.error is None, repr(k.error)[:300]
+    assert t.error is None, repr(t.error)[:300]
+    ok = wait_until(lambda: text(box) == f"1 {n} {n}", 10)
+    assert not helper_errors, repr(helper_errors[0])[:300]
+    assert ok, text(box)
+    rc.close()
+
+
+def test_own_render_loop_is_still_detected():
+    """A component that changes its own state on every render still raises. Passes on
+    40a90a8 (a guard)."""
+
+    @reacton.component
+    def Infinite():
+        state, set_state = reacton.use_state(0)
+        set_state(state + 1)
+        return w.Button(description=str(state))
+
+    rc = core._RenderContext(Infinite(), handle_error=False)
+    finished, k = call_in_thread(lambda: rc.render(rc.element), "K", timeout=TIMEOUT)
+    assert finished, stacks()
+    assert isinstance(k.error, RuntimeError) and "Too many renders triggered" in str(k.error), repr(k.error)
+    finished, _c = call_in_thread(rc.close, "C", timeout=TIMEOUT)
+    assert finished, stacks()
+
+
+def test_renderer_caller_is_not_held_by_a_stream_of_foreign_updates():
+    """K renders because of its own state change. T changes state once during every render
+    pass, until K's call returned (a task that reports progress until the user clicks cancel,
+    a click that K must process). K's call must return without an error, and T's last value
+    (made after K returned) must be rendered. Fails on 40a90a8 ("Too many renders")."""
+    setters = {}
+    streaming, k_returned = threading.Event(), threading.Event()
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        if streaming.is_set() and not k_returned.is_set() and threading.current_thread().name != "T":
+            go.release()
+            done.acquire(timeout=1)  # T reports progress during this pass
+        return w.Button(description=f"{trigger} {progress}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    count = [0]
+
+    def t_target():
+        deadline = time.monotonic() + 40
+        while not k_returned.is_set() and time.monotonic() < deadline:
+            if go.acquire(timeout=0.1):
+                count[0] += 1
+                setters["progress"](count[0])
+                done.release()
+        count[0] += 1
+        setters["progress"](count[0])  # after K returned: must be rendered too
+
+    def k_target():
+        streaming.set()
+        setters["trigger"](1)
+
+    t = Worker(t_target, "T")
+    t.start()
+    finished, k = call_in_thread(k_target, "K", timeout=10)
+    k_returned.set()
+    k.join(45)
+    t.join(45)
+    assert finished, f"K's set() did not return while T kept changing state ({count[0]} updates so far)\n" + stacks()
+    assert k.error is None, repr(k.error)
+    assert t.error is None, repr(t.error)
+    assert wait_until(lambda: text(box) == f"1 {count[0]}", 10), f"lost: shows {text(box)!r}, expected '1 {count[0]}'"
+    rc.close()
+
+
+def test_explicit_render_is_not_starved_by_a_stream():
+    """A holder renders a stream of T's updates (one per pass, until W's render() returned).
+    W calls rc.render() (solara: app.py on a hot reload). It must return.
+    On 40a90a8 K raises "Too many renders" first. With a real scheduler the blocked waiter
+    usually wins the lock when the holder releases it between holds."""
+    setters = {}
+    streaming, w_returned = threading.Event(), threading.Event()
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+
+    @reacton.component
+    def Test(label="old"):
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        if streaming.is_set() and not w_returned.is_set() and threading.current_thread().name != "T":
+            go.release()
+            done.acquire(timeout=1)
+        return w.Button(description=f"{label} {trigger} {progress}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    count = [0]
+
+    def t_target():
+        deadline = time.monotonic() + 40
+        while not w_returned.is_set() and time.monotonic() < deadline:
+            if go.acquire(timeout=0.1):
+                count[0] += 1
+                setters["progress"](count[0])
+                done.release()
+
+    def k_target():
+        streaming.set()
+        setters["trigger"](1)
+
+    t = Worker(t_target, "T")
+    t.start()
+    k = Worker(k_target, "K")
+    k.start()
+    assert wait_until(lambda: count[0] >= 60, 20), "the stream did not start"  # past the hand-over to a helper
+    finished, wt = call_in_thread(lambda: rc.render(Test(label="new")), "W", timeout=10)
+    w_returned.set()
+    wt.join(45)
+    k.join(45)
+    t.join(45)
+    assert finished, f"W's render() waited while another thread kept rendering T's updates ({count[0]})\n" + stacks()
+    assert wait_until(lambda: text(box).startswith("new 1"), 5), text(box)
+    rc.close()
diff --git a/reacton/threads_robustness_test.py b/reacton/threads_robustness_test.py
new file mode 100644
index 0000000..1e31a67
--- /dev/null
+++ b/reacton/threads_robustness_test.py
@@ -0,0 +1,283 @@
+"""Robustness of the render hand-off: an exception in a waiter's batch exit or in a failing
+hold must not leak the render lock or lose a request, an explicit render() is not starved by a
+stream of derived own-state re-renders, and a foreign request consumed by an own pass is not
+a render loop."""
+
+# ruff: noqa: F811  the fixture names shadow the pytest fixture parameters of the same names
+
+import sys
+import threading
+import time
+
+import reacton
+import reacton.ipywidgets as w
+from reacton import core
+
+from ._threads_test_utils import (  # noqa: F401
+    SHORT,
+    TIMEOUT,
+    EveryTimeHook,
+    Injected,
+    Worker,
+    call_in_thread,
+    helper_errors,
+    install,
+    log_hook,
+    source_line,
+    stacks,
+    text,
+    wait_until,
+)
+
+
+def test_exception_in_the_waiters_batch_exit_does_not_leak_the_lock():
+    """W waits in render() while K holds the lock (K is paused in a component body). When K
+    releases, W gets the lock and then leaves its waiting batch. An exception there (a
+    KeyboardInterrupt, a trace function) must not leak the lock or the batch count."""
+    setters = {}
+    k_in_body, k_resume = threading.Event(), threading.Event()
+
+    @reacton.component
+    def Test(label="a"):
+        a, setters["a"] = reacton.use_state(0)
+        if a == 1 and threading.current_thread().name == "K":
+            k_in_body.set()
+            k_resume.wait(TIMEOUT)
+        return w.Button(description=f"{label} {a}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    k = Worker(lambda: setters["a"](1), "K")
+    k.start()
+    assert k_in_body.wait(TIMEOUT)
+    exit_code = core._RenderContext.__exit__.__code__
+    hold_code = core._RenderContext._hold_and_render.__code__
+    fired: list = []
+
+    def trace(frame, event, arg):
+        if not fired and frame.f_code is exit_code and rc.thread_lock.locked() and not rc._is_rendering and k_resume.is_set():
+            fired.append(event)
+            raise Injected()
+        if (
+            not fired
+            and frame.f_code is hold_code
+            and rc.thread_lock.locked()
+            and rc._batch_counter.current() == 0
+            and not rc._is_rendering
+            and k_resume.is_set()
+        ):
+            fired.append(event)
+            raise Injected()
+        return trace
+
+    def w_target():
+        sys.settrace(trace)
+        try:
+            rc.render(Test(label="b"))
+        finally:
+            sys.settrace(None)
+
+    wt = Worker(w_target, "W")
+    wt.start()
+    assert wait_until(lambda: rc._batch_counter.current() == 1, TIMEOUT), "W did not start waiting"
+    k_resume.set()
+    k.join(TIMEOUT)
+    wt.join(TIMEOUT)
+    assert not k.is_alive() and not wt.is_alive(), stacks()
+    assert fired, "the probe did not fire"
+    assert isinstance(wt.error, Injected), repr(wt.error)
+    assert not rc.thread_lock.locked(), "the render lock leaked after an exception in the waiter's batch exit"
+    assert rc._batch_counter.current() == 0, "the waiter's batch count leaked"
+    setters["a"](2)
+    assert wait_until(lambda: text(box).endswith(" 2"), TIMEOUT), text(box)
+    finished, _c = call_in_thread(rc.close, "C", timeout=TIMEOUT)
+    assert finished, "close() hangs on the leaked lock\n" + stacks()
+
+
+def test_request_during_a_failing_hold_is_rendered(log_hook, helper_errors):
+    """handle_error=False. A's pass fails (a component raises); A pauses after its render loop,
+    still holding the lock, before it raises. B asks for a recovery render (B's try fails, B
+    returns). A raises and releases. B's request must still be rendered after the holder raises."""
+    setters = {}
+
+    @reacton.component
+    def Test():
+        fail, setters["fail"] = reacton.use_state(False)
+        if fail:
+            raise ValueError("fail")
+        return w.Button(description=f"ok {fail}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    a_paused, a_resume = threading.Event(), threading.Event()
+
+    def a_target():
+        log_hook.on(threading.current_thread(), "Done with render phase", lambda: (a_paused.set(), a_resume.wait(TIMEOUT * 3)))  # type: ignore[func-returns-value]
+        setters["fail"](True)
+
+    a = Worker(a_target, "A")
+    a.start()
+    assert a_paused.wait(TIMEOUT)
+    finished, b = call_in_thread(lambda: setters["fail"](False), "B", timeout=SHORT * 2)
+    a_resume.set()
+    assert finished and b.error is None, ("B's state change waited for A's render", b.error, stacks())
+    a.join(TIMEOUT)
+    assert isinstance(a.error, ValueError), repr(a.error)
+    assert wait_until(
+        lambda: not rc.thread_lock.locked() and not getattr(rc, "_render_requested", False) and text(box) == "ok False", 3
+    ), f"B's recovery request is still pending: _render_requested={getattr(rc, '_render_requested', None)}, shows {text(box)!r}"
+    assert not helper_errors, helper_errors
+    rc.close()
+
+
+def test_request_after_explicit_render_hold_survives_interrupt(helper_errors):
+    setters = {}
+    arm, b_done = threading.Event(), threading.Event()
+
+    @reacton.component
+    def Test(label="old"):
+        value, setters["value"] = reacton.use_state(0)
+
+        def effect():
+            if arm.is_set():
+                arm.clear()
+
+                def b_target():
+                    setters["value"](1)
+                    b_done.set()
+
+                threading.Thread(target=b_target, name="B", daemon=True).start()
+                b_done.wait(TIMEOUT)
+
+        reacton.use_effect(effect)
+        return w.Button(description=f"{label} {value}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    drain_line = source_line(core._RenderContext.render, "self._drain_render_requests()")
+
+    def trace(frame, event, arg):
+        if event == "line" and frame.f_code is core._RenderContext.render.__code__ and frame.f_lineno == drain_line:
+            raise Injected()
+        return trace
+
+    def a_target():
+        arm.set()
+        sys.settrace(trace)
+        try:
+            rc.render(Test(label="new"))
+        finally:
+            sys.settrace(None)
+
+    a = Worker(a_target, "A")
+    a.start()
+    a.join(TIMEOUT)
+    assert not a.is_alive(), stacks()
+    assert isinstance(a.error, Injected), repr(a.error)
+    assert b_done.is_set(), "the request during A's hold did not run"
+    assert wait_until(
+        lambda: not getattr(rc, "_render_requested", False) and text(box) == "new 1", TIMEOUT
+    ), f"the request stayed pending: _render_requested={getattr(rc, '_render_requested', None)}, shows {text(box)!r}"
+    assert not helper_errors, helper_errors
+    rc.close()
+
+
+def test_explicit_render_is_not_starved_by_derived_own_state():
+    """An effect derives own state from the streamed value, so every pass is followed by an own
+    re-render. Once the render budget hands the stream over to a helper, W's render() must
+    still return."""
+    setters = {}
+    streaming, w_returned = threading.Event(), threading.Event()
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+
+    @reacton.component
+    def Test(label="old"):
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        _derived, set_derived = reacton.use_state(0)
+        reacton.use_effect(lambda: set_derived(progress), [progress])
+        if streaming.is_set() and not w_returned.is_set() and threading.current_thread().name != "T":
+            go.release()
+            done.acquire(timeout=1)
+        return w.Button(description=f"{label} {trigger} {progress}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+    count = [0]
+
+    def t_target():
+        deadline = time.monotonic() + 40
+        while not w_returned.is_set() and time.monotonic() < deadline:
+            if go.acquire(timeout=0.1):
+                count[0] += 1
+                setters["progress"](count[0])
+                done.release()
+
+    def k_target():
+        streaming.set()
+        setters["trigger"](1)
+
+    t = Worker(t_target, "T")
+    t.start()
+    k = Worker(k_target, "K")
+    k.start()
+    assert wait_until(lambda: count[0] >= 150, 30), "the stream did not start"  # past the hand-over to a helper
+    finished, wt = call_in_thread(lambda: rc.render(Test(label="new")), "W", timeout=10)
+    w_returned.set()
+    wt.join(45)
+    k.join(45)
+    t.join(45)
+    assert finished, f"W's render() waited while a helper kept rendering ({count[0]} updates)\n" + stacks()
+    assert wait_until(lambda: text(box).startswith("new 1"), 5), text(box)
+    rc.close()
+
+
+def test_foreign_request_consumed_by_an_own_pass_is_not_a_render_loop(helper_errors):
+    """An effect derives own state from T's value, so the renderer renders own passes. T changes
+    state at the start of every pass (at the log calls "Render phase: " and "Entering nested
+    render phase"). 60 times. That is not a render loop: no "Too many renders", and T's last
+    value is rendered."""
+    n = 60
+    setters = {}
+    go, done = threading.Semaphore(0), threading.Semaphore(0)
+    remaining = [n]
+    armed = threading.Event()
+
+    @reacton.component
+    def Test():
+        trigger, setters["trigger"] = reacton.use_state(0)
+        progress, setters["progress"] = reacton.use_state(0)
+        derived, set_derived = reacton.use_state(0)
+        reacton.use_effect(lambda: set_derived(progress + trigger), [progress, trigger])
+        return w.Button(description=f"{trigger} {progress} {derived}")
+
+    box, rc = reacton.render(Test(), handle_error=False)
+
+    def at_pass_start():
+        if armed.is_set() and remaining[0] > 0:
+            go.release()
+            done.acquire(timeout=TIMEOUT)
+
+    uninstall = install(EveryTimeHook(["Entering nested render phase", "Render phase: "], at_pass_start, skip="T"))
+
+    def t_target():
+        for i in range(n):
+            assert go.acquire(timeout=TIMEOUT)
+            setters["progress"](i + 1)
+            remaining[0] -= 1
+            done.release()
+
+    def k_target():
+        armed.set()
+        setters["trigger"](1)
+
+    try:
+        t = Worker(t_target, "T")
+        t.start()
+        finished, k = call_in_thread(k_target, "K", timeout=60)
+        t.join(60)
+        assert finished, stacks()
+        assert k.error is None, repr(k.error)[:300]
+        assert t.error is None, repr(t.error)[:300]
+        ok = wait_until(lambda: text(box) == f"1 {n} {n + 1}", 10)
+        assert not helper_errors, repr(helper_errors[0])[:300]
+        assert ok, text(box)
+    finally:
+        uninstall()
+    rc.close()