diff --git a/README.rst b/README.rst
index 946b2e8..245c130 100644
--- a/README.rst
+++ b/README.rst
@@ -26,8 +26,7 @@ trame-server can be installed with `pip <https://pypi.org/project/trame-server/>
 
 .. code-block:: bash
 
-    pip install --upgrade trame-server
-
+    pip install trame-server
 
 Usage
 -----------------------------------------------------------
@@ -59,7 +58,7 @@ and are prefixed with ``on_*``.
 * **on_server_exited**    : Trame is exiting its event loop
 
 * **on_error**            : JS error forwarded (msg:str) on the python
-* **on_exception**        : Server exception occurred when interacting with the UI. The exception instance is provided as arg.
+* **on_exception**        : Server exception occurred when interacting with the UI, in a ``@state.change`` listener (sync or async) or in a controller task (``ctrl.add_task``). The exception instance is provided as arg.
 
 * **on_server_reload**    : If callback registered it can be use to hot_reload methods like the UI.
 
@@ -83,3 +82,23 @@ Enjoying trame?
 -----------------------------------------------------------
 
 Share your experience `with a testimonial <https://github.com/Kitware/trame/issues/18>`_ or `with a brand approval <https://github.com/Kitware/trame/issues/19>`_.
+
+Development
+-----------------------------------------------------------
+
+Install the library in dev mode
+
+.. code-block:: bash
+
+    # Create venv and install all dependencies
+    uv sync --all-extras --dev
+
+    # Activate environment
+    source .venv/bin/activate
+
+    # Install commit analysis
+    pre-commit install
+    pre-commit install --hook-type commit-msg
+
+    # Run pre-commit checks
+    pre-commit run --all-files
diff --git a/examples/exceptions/app_trame.py b/examples/exceptions/app_trame.py
index cf8f451..87de6de 100644
--- a/examples/exceptions/app_trame.py
+++ b/examples/exceptions/app_trame.py
@@ -1,9 +1,10 @@
+import asyncio
 import logging
 import sys
 import traceback
 
 from trame.app import TrameApp
-from trame.decorators import life_cycle
+from trame.decorators import change, life_cycle
 from trame.ui.html import DivLayout
 from trame.widgets import html
 
@@ -14,17 +15,25 @@
 class MonitorException(TrameApp):
     def __init__(self, server=None):
         super().__init__(server)
+        self.server.state.test = 1
+        self.server.state.test_async = 1
+        self.server.state.allow_exception = False
         self._build_ui()
-        self.ctrl.on_exception.add(self.on_exception)
 
     def _build_ui(self):
         with DivLayout(self.server) as self.ui:
             html.Button("Divide by 0", click=self.divide_by_zero)
             html.Button("JS error", click="console.error('Yo')")
+            html.Button("Divide by 0 in change listener", click="test += 1")
+            html.Button("Divide by 0 in async change listener", click="test_async += 1")
 
     def divide_by_zero(self):
         return 1 / 0
 
+    @life_cycle.server_ready
+    def on_ready(self, **_):
+        self.state.allow_exception = True
+
     @life_cycle.exception
     def on_exception(self, exception):
         print(">" * 60)
@@ -42,6 +51,19 @@ def on_js_error(self, msg):
 
         print("JS error:", msg)
 
+    @change("test")
+    def test_is_changed(self, allow_exception, **_):
+        if not allow_exception:
+            return 0
+        return 1 / 0
+
+    @change("test_async")
+    async def test_async_is_changed(self, allow_exception, **_):
+        if not allow_exception:
+            return 0
+        await asyncio.sleep(0.1)
+        return 1 / 0
+
 
 def main():
     app = MonitorException()
diff --git a/pyproject.toml b/pyproject.toml
index c7776b3..2d2abed 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -73,6 +73,7 @@ xfail_strict = true
 minversion = "6.0"
 addopts = ["-ra", "--strict-config", "--strict-markers"]
 filterwarnings = ["error"]
+asyncio_default_fixture_loop_scope = "function"
 testpaths = [
     "tests",
 ]
diff --git a/src/trame_server/controller.py b/src/trame_server/controller.py
index 17e89cc..c678335 100644
--- a/src/trame_server/controller.py
+++ b/src/trame_server/controller.py
@@ -65,8 +65,15 @@ class Controller:
     >>> ctrl.on_data_change.clear(set_only=True)  # add, remove, discard, clear
     """
 
-    def __init__(self, translator=None, internal=None, hot_reload=False):
+    def __init__(
+        self, translator=None, internal=None, hot_reload=False, exception_handler=None
+    ):
         super().__setattr__("__trame_hot_reload__", hot_reload)
+        # Child controllers always report to the root controller handler
+        super().__setattr__(
+            "_exception_handler",
+            share(internal, "_exception_handler", exception_handler),
+        )
         super().__setattr__("_translator", translator or Translator())
         super().__setattr__("_triggers", share(internal, "_triggers", {}))
         super().__setattr__(
@@ -387,9 +394,7 @@ def __call__(self, *args, **kwargs):
 
         # Schedule any task
         for task_fn in list(self.task_funcs):
-            results.append(
-                asynchronous.create_task(_safe_call(task_fn, *args, **kwargs))
-            )
+            results.append(self._create_task(_safe_call(task_fn, *args, **kwargs)))
 
         # Figure out return
         if self.func is None:
@@ -400,6 +405,34 @@ def __call__(self, *args, **kwargs):
             return results
         return result
 
+    def _create_task(self, coroutine):
+        task = asynchronous.get_event_loop().create_task(coroutine)
+        task.add_done_callback(self._on_task_done)
+        return task
+
+    def _on_task_done(self, task):
+        if task.cancelled():
+            return
+
+        exception = task.exception()
+        if exception is None:
+            return
+
+        # Never forward failures of on_exception itself to avoid infinite loops
+        handler = self.controller._exception_handler
+        if handler is not None and self.name != "on_exception":
+            try:
+                if handler(exception):
+                    return
+            except Exception:
+                logger.exception("Exception handler failed while reporting")
+
+        logger.error(
+            "Exception raised by %s task",
+            self.name,
+            exc_info=(type(exception), exception, exception.__traceback__),
+        )
+
     def once(self, func):
         """
         Add function to the set of functions to be called when
diff --git a/src/trame_server/core.py b/src/trame_server/core.py
index 529188e..90fa79d 100644
--- a/src/trame_server/core.py
+++ b/src/trame_server/core.py
@@ -103,7 +103,10 @@ def __init__(
         # Shared state + reserve internal keys
         if parent_server is None:
             self._state = State(
-                self.translator, commit_fn=self._push_state, hot_reload=self.hot_reload
+                self.translator,
+                commit_fn=self._push_state,
+                hot_reload=self.hot_reload,
+                exception_handler=self._handle_exception,
             )
             for key in [
                 "scripts",
@@ -128,7 +131,11 @@ def __init__(
 
         # Controller
         if parent_server is None:
-            self._controller = Controller(self.translator, hot_reload=self.hot_reload)
+            self._controller = Controller(
+                self.translator,
+                hot_reload=self.hot_reload,
+                exception_handler=self._handle_exception,
+            )
         else:
             self._controller = Controller(
                 self.translator,
@@ -138,7 +145,11 @@ def __init__(
 
         # Server only context
         if parent_server is None:
-            self._context = State(self.translator, hot_reload=self.hot_reload)
+            self._context = State(
+                self.translator,
+                hot_reload=self.hot_reload,
+                exception_handler=self._handle_exception,
+            )
         else:
             self._context = State(
                 self.translator,
@@ -149,6 +160,20 @@ def __init__(
         # UI (FIXME): use for translator
         self._ui = share(parent_server, "_ui", VirtualNodeManager(self, vn_constructor))
 
+    def _handle_exception(self, exception) -> bool:
+        """
+        Forward an exception raised by a state/context change listener
+        or a controller task to controller.on_exception
+        (i.e. @life_cycle.exception).
+
+        :return: True if the exception was forwarded, False otherwise
+        """
+        on_exception = self.controller.on_exception
+        if on_exception.exists() or on_exception.funcs_once:
+            on_exception(exception)
+            return True
+        return False
+
     def create_child_server(self, translator=None, prefix=None) -> Server:
         translator = translator or Translator(prefix=prefix)
         return Server(translator=translator, parent_server=self)
diff --git a/src/trame_server/state.py b/src/trame_server/state.py
index a5ebd57..d75e852 100644
--- a/src/trame_server/state.py
+++ b/src/trame_server/state.py
@@ -180,8 +180,13 @@ def __init__(
         commit_fn=None,
         hot_reload=False,
         ready=False,
+        exception_handler=None,
     ):
         self._push_state_fn = commit_fn
+        # Child states always report to the root state handler
+        self._exception_handler = share(
+            internal, "_exception_handler", exception_handler
+        )
         self._hot_reload = hot_reload
         self._translator = translator or Translator()
         self._modified_keys = share(internal, "_modified_keys", set())
@@ -444,13 +449,41 @@ def _flush_pending_keys(self) -> set[str]:
             try:
                 coroutine = callback(**reverse_translated_state)
                 if inspect.isawaitable(coroutine):
-                    asynchronous.create_task(coroutine)
+                    task = asynchronous.get_event_loop().create_task(coroutine)
+                    task.add_done_callback(self._on_change_task_done)
             except Exception as e:
-                logger.warning("@change callback exception ignored: %s", e)
+                self._report_exception(e)
 
         self._state_listeners.clear()
         return _keys
 
+    def _report_exception(self, exception):
+        """
+        Forward an exception raised by a @change callback to the exception
+        handler (server.controller.on_exception) when available, otherwise
+        log it so it does not go unnoticed.
+        """
+        if self._exception_handler is not None:
+            try:
+                if self._exception_handler(exception):
+                    return
+            except Exception:
+                logger.exception("Exception handler failed while reporting")
+
+        logger.warning(
+            "@change callback exception ignored: %s",
+            exception,
+            exc_info=(type(exception), exception, exception.__traceback__),
+        )
+
+    def _on_change_task_done(self, task):
+        if task.cancelled():
+            return
+
+        exception = task.exception()
+        if exception is not None:
+            self._report_exception(exception)
+
     def flush(self):
         """
         Force pushing modified state and execute any @state.change listener
diff --git a/tests/test_controller.py b/tests/test_controller.py
index 925be32..c7b32c8 100644
--- a/tests/test_controller.py
+++ b/tests/test_controller.py
@@ -268,3 +268,180 @@ def hr_added_fn(x):
     controller.fn = hr_main_fn
     controller.fn.add(hr_added_fn)
     assert controller.fn(3) == [4, 6]
+
+
+# -----------------------------------------------------------------------------
+# Task exception forwarding
+# -----------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_controller_task_exception_forwarded_to_handler(caplog):
+    from trame_server.controller import Controller  # noqa: PLC0415
+
+    reported = []
+
+    def handler(exception):
+        reported.append(exception)
+        return True
+
+    controller = Controller(exception_handler=handler)
+
+    async def fail():
+        msg = "task boom"
+        raise ValueError(msg)
+
+    controller.fn.add_task(fail)
+    tasks = controller.fn()
+    await asyncio.gather(*tasks, return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert [str(e) for e in reported] == ["task boom"]
+    assert not any(r.levelname == "ERROR" for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_controller_task_exception_logged_without_handler(caplog):
+    from trame_server.controller import Controller  # noqa: PLC0415
+
+    controller = Controller()
+
+    async def fail():
+        raise ValueError()
+
+    controller.fn.add_task(fail)
+    tasks = controller.fn()
+    await asyncio.gather(*tasks, return_exceptions=True)
+    await asyncio.sleep(0)
+
+    records = [
+        r
+        for r in caplog.records
+        if r.levelname == "ERROR" and "Exception raised by fn task" in r.message
+    ]
+    assert len(records) == 1
+    assert records[0].exc_info[0] is ValueError
+
+
+@pytest.mark.asyncio
+async def test_controller_task_exception_logged_when_handler_declines(caplog):
+    from trame_server.controller import Controller  # noqa: PLC0415
+
+    reported = []
+
+    def handler(exception):
+        reported.append(exception)
+        return False
+
+    controller = Controller(exception_handler=handler)
+
+    async def fail():
+        raise ValueError()
+
+    controller.fn.add_task(fail)
+    await asyncio.gather(*controller.fn(), return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert len(reported) == 1
+    assert any(
+        r.levelname == "ERROR" and "Exception raised by fn task" in r.message
+        for r in caplog.records
+    )
+
+
+@pytest.mark.asyncio
+async def test_controller_task_handler_failure_is_logged(caplog):
+    from trame_server.controller import Controller  # noqa: PLC0415
+
+    def handler(exception):
+        raise RuntimeError() from exception
+
+    controller = Controller(exception_handler=handler)
+
+    async def fail():
+        raise ValueError()
+
+    controller.fn.add_task(fail)
+    await asyncio.gather(*controller.fn(), return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert any("Exception handler failed" in r.message for r in caplog.records)
+    assert any("Exception raised by fn task" in r.message for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_controller_task_cancelled_is_ignored(caplog):
+    from trame_server.controller import Controller  # noqa: PLC0415
+
+    reported = []
+    controller = Controller(exception_handler=reported.append)
+
+    async def wait_forever():
+        await asyncio.sleep(10)
+
+    controller.fn.add_task(wait_forever)
+    (task,) = controller.fn()
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert reported == []
+    assert not any(r.levelname == "ERROR" for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_server_controller_task_exception_forwarded_to_on_exception():
+    from trame_server import Server  # noqa: PLC0415
+
+    server = Server()
+    child = server.create_child_server(prefix="child_")
+
+    reported = []
+    server.controller.on_exception.add(reported.append)
+
+    async def fail():
+        msg = "root"
+        raise ValueError(msg)
+
+    async def child_fail():
+        msg = "child"
+        raise ValueError(msg)
+
+    server.controller.fn.add_task(fail)
+    child.controller.fn.add_task(child_fail)
+
+    await asyncio.gather(*server.controller.fn(), return_exceptions=True)
+    await asyncio.gather(*child.controller.fn(), return_exceptions=True)
+    await asyncio.sleep(0)
+
+    assert [str(e) for e in reported] == ["root", "child"]
+
+
+@pytest.mark.asyncio
+async def test_failing_on_exception_task_is_not_forwarded_to_itself(caplog):
+    from trame_server import Server  # noqa: PLC0415
+
+    server = Server()
+    calls = []
+
+    async def failing_on_exception(exception):
+        calls.append(exception)
+        msg = "on_exception failed"
+        raise RuntimeError(msg)
+
+    server.controller.on_exception.add_task(failing_on_exception)
+
+    async def fail():
+        raise ValueError()
+
+    server.controller.fn.add_task(fail)
+    await asyncio.gather(*server.controller.fn(), return_exceptions=True)
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+    # on_exception called once for the original error, never for its own failure
+    assert len(calls) == 1
+    assert isinstance(calls[0], ValueError)
+    assert any(
+        "Exception raised by on_exception task" in r.message for r in caplog.records
+    )
diff --git a/tests/test_state_change.py b/tests/test_state_change.py
index 27b91d0..5ff18e3 100644
--- a/tests/test_state_change.py
+++ b/tests/test_state_change.py
@@ -2,6 +2,10 @@
 
 import pytest
 
+from trame_server import Server
+from trame_server.core import Translator
+from trame_server.state import State
+
 
 @pytest.mark.asyncio
 async def test_change_exception(fake_server):
@@ -55,3 +59,245 @@ def change_a_ok_2(**__kwargs):
     # print(result)
 
     assert expected == result
+
+
+# -----------------------------------------------------------------------------
+# Exception forwarding
+# -----------------------------------------------------------------------------
+
+
+def _ignored_warnings(caplog):
+    return [
+        r
+        for r in caplog.records
+        if r.levelname == "WARNING" and "callback exception ignored" in r.message
+    ]
+
+
+def test_change_exception_forwarded_to_handler(caplog):
+
+    reported = []
+    executed = []
+
+    def handler(exception):
+        reported.append(exception)
+        return True
+
+    state = State(ready=True, exception_handler=handler)
+    state.a = 1
+
+    @state.change("a")
+    def fail(**_):
+        executed.append("fail")
+        msg = "boom"
+        raise ValueError(msg)
+
+    @state.change("a")
+    def ok(**_):
+        executed.append("ok")
+
+    with state:
+        state.a = 2
+
+    assert executed == ["fail", "ok"]
+    assert len(reported) == 1
+    assert isinstance(reported[0], ValueError)
+    assert str(reported[0]) == "boom"
+    assert reported[0].__traceback__ is not None
+    assert _ignored_warnings(caplog) == []
+
+
+def test_change_exception_logged_without_handler(caplog):
+
+    state = State(ready=True)
+    state.a = 1
+
+    @state.change("a")
+    def fail(**_):
+        raise ValueError()
+
+    with state:
+        state.a = 2
+
+    records = _ignored_warnings(caplog)
+    assert len(records) == 1
+    assert records[0].exc_info[0] is ValueError
+
+
+def test_change_exception_logged_when_handler_declines(caplog):
+
+    reported = []
+
+    def handler(exception):
+        reported.append(exception)
+        return False
+
+    state = State(ready=True, exception_handler=handler)
+    state.a = 1
+
+    @state.change("a")
+    def fail(**_):
+        raise ValueError()
+
+    with state:
+        state.a = 2
+
+    assert len(reported) == 1
+    assert len(_ignored_warnings(caplog)) == 1
+
+
+def test_change_exception_handler_failure_does_not_propagate(caplog):
+
+    def handler(exception):
+        raise RuntimeError() from exception
+
+    state = State(ready=True, exception_handler=handler)
+    state.a = 1
+
+    @state.change("a")
+    def fail(**_):
+        raise ValueError()
+
+    with state:
+        state.a = 2
+
+    assert any(
+        r.levelname == "ERROR" and "Exception handler failed" in r.message
+        for r in caplog.records
+    )
+    assert len(_ignored_warnings(caplog)) == 1
+
+
+@pytest.mark.asyncio
+async def test_async_change_exception_forwarded_to_handler(caplog):
+
+    reported = []
+
+    def handler(exception):
+        reported.append(exception)
+        return True
+
+    state = State(ready=True, exception_handler=handler)
+    state.a = 1
+
+    @state.change("a")
+    async def fail(**_):
+        await asyncio.sleep(0)
+        msg = "async boom"
+        raise ValueError(msg)
+
+    with state:
+        state.a = 2
+
+    assert reported == []  # task not executed yet
+    await asyncio.sleep(0.1)
+
+    assert len(reported) == 1
+    assert str(reported[0]) == "async boom"
+    assert _ignored_warnings(caplog) == []
+    assert not any(r.levelname == "ERROR" for r in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_async_change_exception_logged_without_handler(caplog):
+
+    state = State(ready=True)
+    state.a = 1
+
+    @state.change("a")
+    async def fail(**_):
+        raise ValueError()
+
+    with state:
+        state.a = 2
+
+    await asyncio.sleep(0.1)
+    assert len(_ignored_warnings(caplog)) == 1
+
+
+def test_child_state_inherits_exception_handler():
+
+    reported = []
+
+    def handler(exception):
+        reported.append(exception)
+        return True
+
+    root = State(ready=True, exception_handler=handler)
+    child = State(Translator(prefix="child_"), internal=root)
+    child.a = 1
+
+    @child.change("a")
+    def fail(**_):
+        raise ValueError()
+
+    with child:
+        child.a = 2
+
+    assert len(reported) == 1
+
+
+def test_server_change_exception_forwarded_to_on_exception(caplog):
+
+    server = Server()
+    server.state.ready()
+    server.context.ready()
+
+    reported = []
+    server.controller.on_exception.add(reported.append)
+
+    @server.state.change("a")
+    def state_fail(a, **_):
+        if a:
+            msg = "state"
+            raise ValueError(msg)
+
+    @server.context.change("b")
+    def context_fail(b, **_):
+        if b:
+            msg = "context"
+            raise ValueError(msg)
+
+    server.state.a = 1
+    server.state.flush()
+    server.context.b = 1
+    server.context.flush()
+
+    assert [str(e) for e in reported] == ["state", "context"]
+    assert _ignored_warnings(caplog) == []
+
+
+def test_child_server_change_exception_forwarded_to_root_on_exception():
+
+    server = Server()
+    server.state.ready()
+    child = server.create_child_server(prefix="child_")
+
+    reported = []
+    server.controller.on_exception.add(reported.append)
+
+    @child.state.change("a")
+    def fail(a, **_):
+        if a:
+            raise ValueError()
+
+    child.state.a = 1
+    child.state.flush()
+
+    assert len(reported) == 1
+
+
+def test_server_change_exception_logged_without_on_exception(caplog):
+
+    server = Server()
+    server.state.ready()
+
+    @server.state.change("a")
+    def fail(a, **_):
+        if a:
+            raise ValueError()
+
+    server.state.a = 1
+    server.state.flush()
+
+    assert len(_ignored_warnings(caplog)) == 1