Fix container anchor re-engagement during continuous content streaming

Author: Jyotish08

src/textual/_compositor.py

@@ -606,6 +606,7 @@ def add_widget(
                     # Get the region that will be updated
                     sub_clip = clip.intersection(child_region)
 
+                    widget._check_anchor()
                     if widget._anchored and not widget._anchor_released:
                         new_scroll_y = (
                             arrange_result.spatial_map.total_region.bottom
@@ -690,6 +691,7 @@ def add_widget(
                             )
                         layer_order -= 1
                 else:
+                    widget._check_anchor()
                     if widget._anchored and not widget._anchor_released:
                         new_scroll_y = widget.virtual_size.height - (
                             widget.container_size.height

src/textual/widget.py

@@ -825,7 +825,10 @@ def _check_anchor(self) -> None:
         if (
             self._anchored
             and self._anchor_released
-            and self.scroll_y >= self.max_scroll_y
+            and (
+                self.scroll_y >= self.max_scroll_y
+                or self.scroll_target_y >= self.max_scroll_y
+            )
         ):
             self._anchor_released = False
 
@@ -2748,7 +2751,12 @@ def _scroll_to(
         """
 
         if release_anchor:
-            self.release_anchor()
+            if y is not None and y >= self.max_scroll_y:
+                release_anchor = False
+                if self._anchored:
+                    self._anchor_released = False
+            else:
+                self.release_anchor()
         maybe_scroll_x = x is not None and (self.allow_horizontal_scroll or force)
         maybe_scroll_y = y is not None and (self.allow_vertical_scroll or force)
         scrolled_x = scrolled_y = False
@@ -2819,6 +2827,9 @@ def _animate_on_complete() -> None:
             if on_complete is not None:
                 self.call_after_refresh(on_complete)
 
+        if self._anchored and self._anchor_released:
+            self._check_anchor()
+
         return scrolled_x or scrolled_y
 
     @property
@@ -2891,7 +2902,12 @@ def scroll_to(
             The call to scroll is made after the next refresh.
         """
         if release_anchor:
-            self.release_anchor()
+            if y is not None and y >= self.max_scroll_y:
+                release_anchor = False
+                if self._anchored:
+                    self._anchor_released = False
+            else:
+                self.release_anchor()
         animator = self.app.animator
         if x is not None:
             animator.force_stop_animation(self, "scroll_x")
@@ -2908,6 +2924,7 @@ def scroll_to(
                 force=force,
                 on_complete=on_complete,
                 level=level,
+                release_anchor=release_anchor,
             )
         else:
             self.call_after_refresh(
@@ -2921,6 +2938,7 @@ def scroll_to(
                 force=force,
                 on_complete=on_complete,
                 level=level,
+                release_anchor=release_anchor,
             )
 
     def scroll_relative(

tests/test_anchor.py

@@ -0,0 +1,45 @@
+from textual.app import App, ComposeResult
+from textual.containers import VerticalScroll
+from textual.widgets import Static
+
+
+async def test_anchor_streaming_repro_flow() -> None:
+    class AnchorStuck(App):
+        def compose(self) -> ComposeResult:
+            with VerticalScroll(id="v"):
+                for i in range(30):
+                    yield Static(f"line {i}")
+
+    app = AnchorStuck()
+    async with app.run_test() as pilot:
+        v = app.query_one("#v", VerticalScroll)
+        v.anchor()
+        await pilot.pause()
+
+        assert v.is_anchored
+        assert not v._anchor_released
+        assert v.scroll_y == v.max_scroll_y
+
+        # User scrolls up by 1 line while streaming
+        v.scroll_relative(y=-1, animate=False)
+        await pilot.pause()
+        assert v._anchor_released is True
+        frozen_scroll_y = v.scroll_y
+
+        # Mounting items while released keeps scroll position frozen
+        v.mount(Static("new line 0"))
+        await pilot.pause()
+        assert v._anchor_released is True
+        assert v.scroll_y == frozen_scroll_y
+
+        # User scrolls back down to bottom edge
+        v.scroll_to(y=v.max_scroll_y, animate=False)
+        await pilot.pause()
+        assert v._anchor_released is False
+        assert v.scroll_y == v.max_scroll_y
+
+        # Mounting new items after re-engaging keeps viewport glued to bottom
+        v.mount(Static("new line 1"))
+        await pilot.pause()
+        assert v._anchor_released is False
+        assert v.scroll_y == v.max_scroll_y