Stage 3 review: route the real abandonment through the owner, and mar…

…k stale requests Cleanup failure -- the one path that actually abandons reserved emission -- stopped emission by setting the flag itself, so the owner was never told: the rows stayed withheld and the renderer stayed bound. It goes through the same door as every other abandonment now. The later call that would have notified could not help, since it finds emission already stopped and returns. Emptying the pending cursor-report queue could not discard the replies: they are already in the terminal's hands. The next one to arrive was then matched against whatever request came after the clear -- the oldest reply answering the newest question, publishing an origin from a screen that no longer exists. The entries stay in the queue now, marked, so each reply is still consumed in order and each one is refused.
tleonhardt · Sep 12, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026 · Sep 10, 2026
commit b01eeabb2494140f1b5dd2941af0d312e05b8443
@@ -123,7 +123,10 @@ def __init__(self, renderer: "Renderer", display: "TerminalDisplay", lock: Termi
         self._pending_error: BaseException | None = None
         self._prompt_anchor: int | None = None
         self._resynchronization_reason: str | None = None
-        self._pending_cpr: deque[Generations] = deque()
+        # One entry per outstanding request, in the order they went out. An entry is ``None``
+        # 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.
@@ -420,7 +423,7 @@ def _clear_through_bridge(self) -> None:
                 self._originals["clear"]()
             finally:
                 self._prompt_anchor = None
-                self._discard_pending_cursor_reports()
+                self._invalidate_pending_cursor_reports()
                 self.require_resynchronization("the renderer cleared the screen")
 
     # -- prepare and commit ----------------------------------------------------------------
@@ -544,8 +547,11 @@ def _attempt_cleanup(self) -> bool:
             output.flush()
             self._display.reconfigure()
         except Exception as error:  # noqa: BLE001 - cleanup failing is itself the answer
-            self._reserved_emission_stopped = True
-            self._pending_error = error
+            # Through the same door as every other abandonment, so the owner is told and can
+            # give the rows back. Setting the flag here directly would stop emission while
+            # leaving the reservation installed and the renderer bound -- and the later call
+            # that would have notified now returns early, having found it already stopped.
+            self.stop_reserved_emission(error)
             return False
         return True
 
@@ -725,7 +731,9 @@ def report_cursor_row(self, row: int) -> bool:
 
         Replies carry no generation on the wire, so they are correlated by order against the
         requests this bridge made. A reply from before a geometry change describes a screen
-        that no longer exists and must not satisfy the request made after it.
+        that no longer exists and must not satisfy the request made after it. Requests whose
+        screen has since been cleared away are kept in the queue but marked: their replies are
+        still coming and still have to be consumed in order, and none of them can be believed.
 
         A reply is stale when anything about the terminal has changed since the request went
         out -- a resize, an owner change, or managed output that moved the cursor the terminal
@@ -757,7 +765,7 @@ def report_cursor_row(self, row: int) -> bool:
             # Popped whatever the outcome: replies correlate by order, so dropping one without
             # taking it off the queue would answer every later request with its predecessor.
             generations = self._pending_cpr.popleft()
-            if generations != self.generations():
+            if generations is None or generations != self.generations():
                 self._settle_renderer_cpr()
                 return False
 
@@ -771,15 +779,16 @@ def report_cursor_row(self, row: int) -> bool:
             self._renderer.report_absolute_cursor_row(row)
             return True
 
-    def _discard_pending_cursor_reports(self) -> None:
-        """Drop every outstanding request, settling the bookkeeping each one owns.
+    def _invalidate_pending_cursor_reports(self) -> None:
+        """Mark every outstanding request unbelievable, without forgetting that it is coming.
 
-        Used where the screen changed underneath the requests themselves. Left in the queue,
-        the next reply to arrive would be matched to a request made about a different screen.
+        Used where the screen changed underneath the requests themselves. Emptying the queue
+        would not stop the replies: they are already in the terminal's hands, and the next one
+        to arrive would be matched against whatever request came *after* the change -- the
+        oldest reply answering the newest question. The entries stay, marked, so each reply is
+        still consumed in order and each one is refused.
         """
-        while self._pending_cpr:
-            self._pending_cpr.popleft()
-            self._settle_renderer_cpr()
+        self._pending_cpr = deque([None] * len(self._pending_cpr))
 
     def _settle_renderer_cpr(self) -> None:
         """Resolve one of the renderer's own pending reports, if it has any.

@@ -1196,3 +1196,48 @@ def test_abandoning_emission_twice_notifies_once(self) -> None:
         harness.bridge.stop_reserved_emission(OSError("and again"))
         assert released == [1]
         assert str(harness.bridge.take_pending_error()) == "terminal went away"
+
+    def test_a_failed_cleanup_tells_the_owner_to_release(self) -> None:
+        """Review finding: the one path that really abandons emission skipped the transition."""
+        released: list[int] = []
+        harness = self.bound()
+        harness.bridge.set_emission_stopped_handler(lambda: released.append(1))
+
+        prepared = harness.bridge.prepare(harness.app)
+        assert prepared is not None
+        harness.backend.stdout = AlwaysFailingTtyStream()
+        assert harness.bridge.commit(prepared) is False
+
+        assert harness.bridge.reserved_emission_stopped is True
+        assert released == [1]
+
+    def test_a_reply_in_transit_when_the_screen_cleared_is_not_reused(self) -> None:
+        """Review finding: emptying the queue lets the next reply answer the wrong request."""
+        harness = self.bound()
+        harness.renderer.request_absolute_cursor_position = lambda: None  # type: ignore[method-assign]
+        harness.bridge.set_prompt_anchor(7)
+
+        harness.bridge.request_cursor_position()  # request A, about the pre-clear screen
+        with set_app(harness.app):
+            harness.renderer.clear()
+        harness.bridge.request_cursor_position()  # request B, about the cleared screen
+
+        # Reply A arrives late. It describes the screen before the clear.
+        assert harness.bridge.report_cursor_row(9) is False
+        assert harness.bridge.prompt_anchor is None
+
+        # Reply B is the one that establishes the origin.
+        assert harness.bridge.report_cursor_row(4) is True
+        assert harness.bridge.prompt_anchor == 4
+
+    def test_the_renderers_own_bookkeeping_is_settled_for_each_stale_reply(self) -> None:
+        harness = self.bound()
+        harness.renderer.request_absolute_cursor_position = lambda: None  # type: ignore[method-assign]
+        harness.bridge.request_cursor_position()
+        pending: Future[None] = Future()
+        harness.renderer._waiting_for_cpr_futures.append(pending)
+        with set_app(harness.app):
+            harness.renderer.clear()
+
+        assert harness.bridge.report_cursor_row(9) is False
+        assert pending.done() is True
@@ -739,3 +739,26 @@ def test_the_failure_is_still_reported(self) -> None:
             assert isinstance(harness.toolbar.take_pending_error(), OSError)
         finally:
             harness.close()
+
+    def test_a_failed_cleanup_releases_the_rows(self) -> None:
+        """The path that truly abandons emission has to reach the owner like any other.
+
+        Driven at the cleanup itself: a real partial commit needs a running event loop to
+        render a prompt session, and what is under test here is the wiring from "cleanup
+        failed" to "the owner gave the rows back", not how the commit got there.
+        """
+        harness = Harness()
+        try:
+            original_render = harness.app.renderer.render
+            harness.toolbar.start()
+            bridge = harness.toolbar.bridge
+            assert bridge is not None
+
+            harness.backend.stdout = AlwaysFailingStream()
+            assert bridge._attempt_cleanup() is False
+
+            assert bridge.reserved_emission_stopped is True
+            assert harness.toolbar.is_active is False
+            assert harness.app.renderer.render == original_render
+        finally:
+            harness.close()