feat: add reactivity_loss_strategy option

Author: CNSeniorious000

packages/hmr/reactivity/context.py

@@ -34,14 +34,18 @@ def enter(self, computation: BaseComputation):
                 computation.dependencies.update(old_dependencies)
             raise
         else:
-            if not computation.dependencies:
+            if not computation.dependencies and (strategy := computation.reactivity_loss_strategy) != "ignore":
+                if strategy == "restore":
+                    for dep in old_dependencies:
+                        dep.subscribers.add(computation)
+                    computation.dependencies.update(old_dependencies)
+                    return
                 from pathlib import Path
                 from sysconfig import get_path
                 from warnings import warn
 
-                # in the future this may be a configurable behavior (e.g., restoring the previous dependencies)
                 msg = "lost all its dependencies" if old_dependencies else "has no dependencies"
-                warn(f"Computation {computation} {msg} and will never be auto-triggered.", RuntimeWarning, skip_file_prefixes=(str(Path(__file__).parent), str(Path(get_path("stdlib")).resolve())))
+                warn(f"{computation} {msg} and will never be auto-triggered.", RuntimeWarning, skip_file_prefixes=(str(Path(__file__).parent), str(Path(get_path("stdlib")).resolve())))
         finally:
             last = self.current_computations.pop()
             assert last is computation  # sanity check

packages/hmr/reactivity/hmr/utils.py

@@ -71,6 +71,7 @@ def wrapper() -> T:
         return functions[key]()
 
     memo = Derived(wrapper, context=HMR_CONTEXT)
+    memo.reactivity_loss_strategy = "ignore"  # Manually invalidated on source change, so reactivity loss is safe to ignore
     memos[key] = memo, source
 
     return _return(wraps(func)(memo))

packages/hmr/reactivity/primitives.py

@@ -1,5 +1,5 @@
 from collections.abc import Callable
-from typing import Any, Self, overload
+from typing import Any, Literal, Self, overload
 from weakref import WeakSet
 
 from .context import Context, default_context
@@ -75,6 +75,18 @@ def trigger(self) -> Any: ...
     def __call__(self) -> T:
         return self.trigger()
 
+    reactivity_loss_strategy: Literal["ignore", "warn", "restore"] = "warn"
+    """
+    A computation without dependencies usually indicates a code mistake.
+    ---
+    By default, a warning is issued when a computation completes without collecting any dependencies.
+    This often happens when signal access is behind non-reactive conditions or caching.
+    You can set this to `"restore"` to automatically preserve previous dependencies as a **temporary workaround**.
+    The correct practice is to replace those conditions with reactive ones (e.g. `Signal`) or use `Derived` for caching.
+    * * *
+    Consider `"ignore"` only when extending this library and manually managing dependencies. Use with caution.
+    """
+
 
 class Signal[T](Subscribable):
     def __init__(self, initial_value: T = None, check_equality=True, *, context: Context | None = None):

tests/py/test_hmr.py

@@ -189,7 +189,7 @@ def f(): pass
     assert recwarn.list == []
 
 
-def test_cache_across_reloads_with_other_decorators(recwarn: pytest.WarningsRecorder):
+def test_cache_across_reloads_with_other_decorators():
     with environment() as env:
         env["main.py"] = """
                 from reactivity.hmr.utils import cache_across_reloads
@@ -203,11 +203,6 @@ def two(): return 2
         assert env.stdout_delta == "3\n3\n1\n"  # inner function being called twice, while the outer one only once
         assert ns["two"] == 2
 
-        warning = recwarn.pop(RuntimeWarning)
-        assert warning.lineno == 4  # f()
-        assert warning.filename == "main.py"
-    assert recwarn.list == []
-
 
 def test_cache_across_reloads_cache_lifespan():
     with environment() as env: