Stage 3 review: an unbound bridge passes calls through instead of bre…

…aking them Unbinding leaves a wrapper somebody else installed over ours in place, because it is not ours to remove -- and that wrapper goes on calling in here afterwards. Clearing the saved originals at the same time left those calls with nothing to delegate to: a render that emitted nothing, and a fired after-render event that raised on a None. The originals are kept now, and every intercepted method checks whether the bridge is still bound. Unbound it is not the terminal's owner, so it passes the call straight through to upstream rather than taking the lock, preparing a frame, or withholding a notification it has no business withholding. The earlier restoration tests used standalone replacements, which never called back in, so they could not see this. The new ones delegate.
tleonhardt · Sep 12, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026
commit 8f745a286b5ac877f27727136f9541b4028f0e47
@@ -127,10 +127,12 @@ def __init__(self, renderer: "Renderer", display: "TerminalDisplay", lock: Termi
         # once the screen it asked about is gone: the reply is still coming, so the place has
         # to be kept, but nothing it says can be believed.
         self._pending_cpr: deque[Generations | None] = deque()
-        # The renderer methods this bridge replaced, by name, empty while unbound. Kept so
-        # preparation can call the real render: calling the attribute would re-enter the
-        # wrapper and never terminate.
+        # The upstream methods this bridge wraps, by name. Kept after unbinding as well as
+        # during: preparation calls the real render through this rather than the attribute,
+        # which would re-enter the wrapper and never terminate -- and a wrapper somebody else
+        # installed over ours may outlive the binding and still call in here.
         self._originals: dict[str, Any] = {}
+        self._bound = False
         # What this bridge put in their place, so teardown can tell its own replacements from
         # something another caller installed afterwards.
         self._installed: dict[str, Any] = {}
@@ -355,7 +357,7 @@ def bind(self, app: "Application[Any]") -> None:
 
         :param app: the application whose renders are being intercepted
         """
-        if self._originals:
+        if self._bound:
             return
         renderer = self._renderer
         replacements = {
@@ -373,6 +375,7 @@ def bind(self, app: "Application[Any]") -> None:
         self._originals = {name: getattr(renderer, name) for name in replacements}
         self._installed = dict(replacements)
         self._bound_app = app
+        self._bound = True
         # Upstream fires this after ``render()`` returns, whatever the wrapper decided to do,
         # so a frame the bridge skipped would still tell everything waiting on a rendered
         # frame that one had happened -- including the command display's readiness signal.
@@ -398,12 +401,11 @@ def unbind(self) -> None:
         event, self._after_render_event = self._after_render_event, None
         if event is not None and getattr(event, "fire", None) == self._after_render_installed:
             setattr(event, "fire", self._after_render_original)  # noqa: B010
-        self._after_render_original = None
         self._after_render_installed = None
 
-        originals, self._originals = self._originals, {}
+        self._bound = False
         installed, self._installed = self._installed, {}
-        for name, original in originals.items():
+        for name, original in self._originals.items():
             # Restored only where this bridge's replacement is still in place. Another caller
             # may have wrapped the renderer since -- for tracing, for a test -- and putting
             # the original back over theirs would silently undo it.
@@ -414,10 +416,19 @@ def unbind(self) -> None:
     def _render_through_bridge(self, app: "Application[Any]", layout: Any, is_done: bool = False) -> None:
         """Prepare and commit one frame, telling anything waiting that an attempt was made.
 
+        Unbound, this passes straight through. A wrapper installed over this one -- for
+        tracing, for a test -- is left in place by :meth:`unbind` precisely because it is not
+        ours to remove, and it goes on calling in here afterwards. The bridge is no longer the
+        terminal's owner then, so the honest answer is upstream's own behaviour rather than an
+        error.
+
         :param app: the application being rendered
         :param layout: the layout to render; upstream passes ``app.layout``
         :param is_done: whether this is the final frame of a prompt
         """
+        if not self._bound:
+            self._originals["render"](app, layout, is_done)
+            return
         try:
             self._render_frame(app, layout, is_done)
         finally:
@@ -470,7 +481,7 @@ def _fire_after_render_through_bridge(self) -> None:
         the screen: layout metadata is published from it, and the command display treats it as
         the signal that its first frame has been drawn.
         """
-        if not self._last_emission_committed:
+        if self._bound and not self._last_emission_committed:
             return
         self._after_render_original()
 
@@ -483,6 +494,9 @@ def _request_cursor_position_through_bridge(self) -> None:
         actually started waiting for.
         """
         renderer = self._renderer
+        if not self._bound:
+            self._originals["request_absolute_cursor_position"]()
+            return
         with self._lock.transaction("cursor position request"):
             if self._reserved_emission_stopped:
                 return
@@ -497,6 +511,9 @@ def _report_cursor_row_through_bridge(self, row: int) -> None:
 
         :param row: the one-based physical row the terminal reported
         """
+        if not self._bound:
+            self._originals["report_absolute_cursor_row"](row)
+            return
         self.report_cursor_row(row)
 
     def _erase_through_bridge(self, leave_alternate_screen: bool = True) -> None:
@@ -508,6 +525,9 @@ def _erase_through_bridge(self, leave_alternate_screen: bool = True) -> None:
 
         :param leave_alternate_screen: passed through to upstream
         """
+        if not self._bound:
+            self._originals["erase"](leave_alternate_screen)
+            return
         self._last_emission_committed = False
         with self._lock.transaction("erase"):
             try:

@@ -1433,3 +1433,121 @@ def test_an_event_replaced_while_bound_is_left_alone(self) -> None:
         harness.app.after_render.fire = replacement  # type: ignore[method-assign]
         harness.bridge.unbind()
         assert harness.app.after_render.fire is replacement
+
+
+class TestDelegatingWrappers:
+    """Someone else's wrapper may outlive the bridge and still call into it."""
+
+    @pytest.fixture(autouse=True)
+    def _event_loop(self) -> Any:
+        """Upstream builds an asyncio Future per cursor request, which needs a loop."""
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            yield
+        finally:
+            asyncio.set_event_loop(None)
+            loop.close()
+
+    def bound(self) -> Harness:
+        harness = Harness()
+        harness.stream_recorder = RecordingTtyStream()
+        harness.backend.stdout = harness.stream_recorder
+        harness.renderer.cpr_support = CPR_Support.SUPPORTED
+        harness.bridge.bind(harness.app)
+        return harness
+
+    def test_a_wrapper_delegating_to_the_bridge_still_renders_after_unbinding(self) -> None:
+        """Review finding: the wrapper is kept, so what it delegates to has to keep working."""
+        harness = self.bound()
+        calls: list[int] = []
+        delegate = harness.renderer.render
+
+        def tracing_render(*args: Any, **kwargs: Any) -> None:
+            calls.append(1)
+            delegate(*args, **kwargs)
+
+        harness.renderer.render = tracing_render  # type: ignore[method-assign]
+        harness.bridge.unbind()
+
+        harness.stream_recorder.truncate(0)
+        harness.stream_recorder.seek(0)
+        with set_app(harness.app):
+            harness.renderer.render(harness.app, harness.app.layout)
+
+        assert calls == [1]
+        assert "hello" in harness.stream_recorder.getvalue()
+
+    def test_an_unbound_bridge_renders_without_taking_the_terminal(self) -> None:
+        """It is not the terminal's owner any more, so it passes the call straight through."""
+        harness = self.bound()
+        delegate = harness.renderer.render
+
+        def passing_render(*args: Any, **kwargs: Any) -> None:
+            delegate(*args, **kwargs)
+
+        harness.renderer.render = passing_render  # type: ignore[method-assign]
+        harness.bridge.unbind()
+
+        harness.stream_recorder.truncate(0)
+        harness.stream_recorder.seek(0)
+        with set_app(harness.app):
+            harness.renderer.render(harness.app, harness.app.layout)
+        assert all(state is None for state in harness.stream_recorder.transactions)
+
+    def test_a_wrapper_delegating_to_the_bridge_can_still_fire_after_render(self) -> None:
+        harness = self.bound()
+        fired: list[int] = []
+        delegate = harness.app.after_render.fire
+
+        def tracing_fire() -> None:
+            fired.append(1)
+            delegate()
+
+        harness.app.after_render.fire = tracing_fire  # type: ignore[method-assign]
+        harness.bridge.unbind()
+
+        handled: list[int] = []
+        harness.app.after_render += lambda _app: handled.append(1)
+        harness.app.after_render.fire()
+
+        assert fired == [1]
+        assert handled == [1]
+
+    def test_delegated_erase_and_clear_still_work_after_unbinding(self) -> None:
+        harness = self.bound()
+        harness.renderer.request_absolute_cursor_position = lambda: None  # type: ignore[method-assign]
+        erase, clear = harness.renderer.erase, harness.renderer.clear
+
+        def passing_erase(*args: Any, **kwargs: Any) -> None:
+            erase(*args, **kwargs)
+
+        def passing_clear() -> None:
+            clear()
+
+        harness.renderer.erase = passing_erase  # type: ignore[method-assign]
+        harness.renderer.clear = passing_clear  # type: ignore[method-assign]
+        harness.bridge.unbind()
+
+        with set_app(harness.app):
+            harness.renderer.erase()
+            harness.renderer.clear()
+
+    def test_delegated_cursor_reports_still_work_after_unbinding(self) -> None:
+        harness = self.bound()
+        request = harness.renderer.request_absolute_cursor_position
+        report = harness.renderer.report_absolute_cursor_row
+
+        def passing_request() -> None:
+            request()
+
+        def passing_report(row: int) -> None:
+            report(row)
+
+        harness.renderer.request_absolute_cursor_position = passing_request  # type: ignore[method-assign]
+        harness.renderer.report_absolute_cursor_row = passing_report  # type: ignore[method-assign]
+        harness.bridge.unbind()
+
+        harness.renderer.request_absolute_cursor_position()
+        harness.renderer.report_absolute_cursor_row(4)
+        assert harness.renderer._min_available_height == 23 - 4 + 1