Fix cleanup of cross-object reactive watchers

src/textual/reactive.py

@@ -177,15 +177,40 @@ def __rich_repr__(self) -> rich.repr.Result:
 
     @classmethod
     def _clear_watchers(cls, obj: Reactable) -> None:
-        """Clear any watchers on a given object.
+        """Clear any watchers created by or attached to a given object.
 
         Args:
             obj: A reactive object.
         """
-        try:
-            getattr(obj, "__watchers").clear()
-        except AttributeError:
-            pass
+        watchers: dict[str, list[tuple[Reactable, WatchCallbackType]]]
+        watchers = getattr(obj, "__watchers", {})
+        for attribute_name, watcher_list in watchers.items():
+            for watcher, _callback in watcher_list:
+                watching: list[tuple[Reactable, str]]
+                watching = getattr(watcher, "__watching", [])
+                watching[:] = [
+                    (watched, watched_attribute)
+                    for watched, watched_attribute in watching
+                    if watched is not obj or watched_attribute != attribute_name
+                ]
+        watchers.clear()
+
+        watching = getattr(obj, "__watching", [])
+        for watched, attribute_name in watching:
+            watched_watchers: dict[str, list[tuple[Reactable, WatchCallbackType]]] = (
+                getattr(watched, "__watchers", {})
+            )
+            watcher_list = watched_watchers.get(attribute_name)
+            if watcher_list is None:
+                continue
+            watcher_list[:] = [
+                (watcher, callback)
+                for watcher, callback in watcher_list
+                if watcher is not obj
+            ]
+            if not watcher_list:
+                watched_watchers.pop(attribute_name, None)
+        watching.clear()
 
     @property
     def owner(self) -> Type[MessageTarget]:
@@ -239,13 +264,13 @@ def _initialize_object(cls, obj: Reactable) -> None:
             reactive._initialize_reactive(obj, name)
 
     @classmethod
-    def _reset_object(cls, obj: object) -> None:
+    def _reset_object(cls, obj: Reactable) -> None:
         """Reset reactive structures on object (to avoid reference cycles).
 
         Args:
             obj: A reactive object.
         """
-        getattr(obj, "__watchers", {}).clear()
+        cls._clear_watchers(obj)
         getattr(obj, "__computes", []).clear()
 
     def __set_name__(self, owner: Type[MessageTarget], name: str) -> None:
@@ -530,3 +555,13 @@ def _watch(
         current_value = getattr(obj, attribute_name, None)
         invoke_watcher(obj, callback, current_value, current_value)
     watcher_list.append((node, callback))
+    watching: list[tuple[Reactable, str]] | None
+    watching = getattr(node, "__watching", None)
+    if watching is None:
+        watching = []
+        setattr(node, "__watching", watching)
+    if not any(
+        watched is obj and watched_attribute == attribute_name
+        for watched, watched_attribute in watching
+    ):
+        watching.append((obj, attribute_name))

tests/test_reactive.py

@@ -646,6 +646,62 @@ def callback(self) -> None:
         assert counter == 2
 
 
+async def test_external_watcher_removed_with_owner() -> None:
+    """External watchers should not retain a node after it is removed."""
+
+    callback_count = 0
+
+    class Watcher(Widget):
+        def on_mount(self) -> None:
+            self.watch(self.app, "value", self.on_value)
+
+        def on_value(self) -> None:
+            nonlocal callback_count
+            callback_count += 1
+
+    class WatchApp(App[None]):
+        value = reactive(0)
+
+        def compose(self) -> ComposeResult:
+            yield Watcher()
+
+    app = WatchApp()
+    async with app.run_test():
+        watcher = app.query_one(Watcher)
+        assert callback_count == 1
+        assert getattr(app, "__watchers")["value"]
+
+        await watcher.remove()
+
+        assert not getattr(app, "__watchers").get("value")
+        assert not getattr(watcher, "__watching")
+        app.value = 1
+        assert callback_count == 1
+
+
+async def test_external_watcher_tracking_removed_with_observed_node() -> None:
+    """Removing an observed node should clear the owner's tracking entry."""
+
+    class Observed(Widget):
+        value = reactive(0)
+
+    class WatchApp(App[None]):
+        def compose(self) -> ComposeResult:
+            yield Observed()
+
+        def on_mount(self) -> None:
+            self.watch(self.query_one(Observed), "value", lambda: None)
+
+    app = WatchApp()
+    async with app.run_test():
+        observed = app.query_one(Observed)
+        assert (observed, "value") in getattr(app, "__watching")
+
+        await observed.remove()
+
+        assert (observed, "value") not in getattr(app, "__watching")
+
+
 async def test_external_watch_init_does_not_propagate() -> None:
     """Regression test for https://github.com/Textualize/textual/issues/3878.