Do not broadcast anywidget echo_update acks back to clients (#10178)

Closes #10173, MO-6686

Since 0.23.12 (#9454), the kernel coerced ipywidget's `echo_update` into
an ordinary update and broadcast it, feeding a client its own write
back. These messages re-fired `change:` listeners for traits that did
not change and also looped widgets that call `save_changes` in such a
listener.

The echo existed only to get frontend-driven trait changes into the
server replay state (#9420). Every client write already reaches the
server as a `ModelCommand`, so `SessionView` now records it into replay
state directly (stripping `_esm`/`_css`, since replayed state is served
to future viewers), and the echo is dropped.

Author: manzt

marimo/_plugins/ui/_impl/comm.py

@@ -113,13 +113,10 @@ def _create_model_message(
 ) -> ModelMessage | None:
     """Create the appropriate ModelMessage based on the method field.
 
-    Returns None for unknown methods that should be skipped. `data` is
-    the ipywidgets-shaped comm payload; `esm_spec` is minted by the
-    comm itself, never supplied by comm callers.
-
-    `echo_update` is coerced to `ModelUpdate`: marimo has no echo
-    protocol, and dropping echoes would lose frontend-driven trait
-    changes from reconnect replay.
+    Returns None for methods that should be skipped, including
+    `echo_update`. `data` is the ipywidgets-shaped comm payload;
+    `esm_spec` is minted by the comm itself, never supplied by comm
+    callers.
     """
     bbuffers = [_ensure_bytes(b) for b in buffers]
     method = data.get("method", "update")
@@ -146,14 +143,11 @@ def _create_model_message(
             buffers=bbuffers,
         )
     elif method == "echo_update":
-        # Preserve frontend-driven trait changes for reconnect replay.
-        # anywidget/ipywidgets can emit echo_update as the synchronisation
-        # acknowledgement path; dropping it causes stale replay state.
-        return ModelUpdate(
-            state=state,
-            buffer_paths=buffer_paths,
-            buffers=bbuffers,
-        )
+        # ipywidgets' acknowledgement of a client write. Broadcasting
+        # it would feed the write back into the sender's model,
+        # re-firing change listeners. Reconnect replay records client
+        # writes server-side instead (SessionView.add_control_request).
+        return None
     else:
         LOGGER.warning("Unknown method: %s, skipping", method)
         return None

marimo/_session/state/session_view.py

@@ -40,6 +40,8 @@
     CreateNotebookCommand,
     ExecuteCellCommand,
     ExecuteCellsCommand,
+    ModelCommand,
+    ModelUpdateMessage,
     SyncGraphCommand,
     UpdateUIElementCommand,
 )
@@ -229,6 +231,37 @@ def add_control_request(self, request: CommandMessage) -> None:
                 self._add_ui_value(object_id, value)
             for execution_request in request.execution_requests:
                 self._add_last_run_code(execution_request)
+        elif isinstance(request, ModelCommand):
+            self._apply_model_command(request)
+
+    def _apply_model_command(self, request: ModelCommand) -> None:
+        """Merge a client's model write into replay state.
+
+        The kernel does not echo client writes back (see
+        `_create_model_message` in `marimo._plugins.ui._impl.comm`), so
+        the command itself is the only record of frontend-driven trait
+        changes for reconnect replay.
+        """
+        message = request.message
+        if not isinstance(message, ModelUpdateMessage):
+            # Custom messages are ephemeral — never replayed.
+            return
+        view = self.model_states.get(request.model_id)
+        if view is None:
+            return
+        # Clients may write widget state, never code or style: replayed
+        # state reaches future viewers. Mirrors the kernel-side filter
+        # in MarimoCommManager.receive_comm_message.
+        state = {
+            k: v for k, v in message.state.items() if k not in ("_esm", "_css")
+        }
+        view.apply_update(
+            ModelUpdate(
+                state=state,
+                buffer_paths=message.buffer_paths,
+                buffers=request.buffers,
+            )
+        )
 
     def add_stdin(self, stdin: str) -> None:
         self._touch()

tests/_plugins/ui/_impl/test_anywidget.py

@@ -297,6 +297,55 @@ class TestWidget(_anywidget.AnyWidget):
         assert ui_element.value == {"value": 42}
         assert ui_element.widget.value == 42
 
+    @staticmethod
+    async def test_client_update_not_echoed_back() -> None:
+        """Client write → observer update is broadcast, the write is not.
+
+        A frontend change runs a Python observer that regenerates a
+        chart trait; only the observer's kernel-driven update may go
+        back out to clients.
+        """
+        from unittest.mock import patch
+
+        from marimo._plugins.ui._impl.anywidget.init import (
+            WIDGET_COMM_MANAGER,
+        )
+
+        class ChartWidget(_anywidget.AnyWidget):
+            _esm = ""
+            param = traitlets.Int(0).tag(sync=True)
+            chart = traitlets.Unicode("").tag(sync=True)
+
+            @traitlets.observe("param")
+            def _redraw(self, change) -> None:
+                self.chart = f"<svg>{change['new']}</svg>"
+
+        w = anywidget(ChartWidget())
+        model_id = WidgetModelId(w.widget._model_id)
+
+        with patch(
+            "marimo._plugins.ui._impl.comm.broadcast_notification"
+        ) as mock_broadcast:
+            WIDGET_COMM_MANAGER.receive_comm_message(
+                ModelCommand(
+                    model_id=model_id,
+                    message=ModelUpdateMessage(
+                        state={"param": 3}, buffer_paths=[]
+                    ),
+                    buffers=[],
+                )
+            )
+
+        assert w.widget.param == 3
+        updates = [
+            call.args[0].message.state
+            for call in mock_broadcast.call_args_list
+        ]
+        # The observer's chart refresh is broadcast...
+        assert {"chart": "<svg>3</svg>"} in updates
+        # ...the echo of the client's own write is not.
+        assert all("param" not in state for state in updates)
+
     @staticmethod
     async def test_buffers() -> None:
         class BufferWidget(_anywidget.AnyWidget):

tests/_plugins/ui/_impl/test_comm.py

@@ -168,19 +168,51 @@ def test_comm_broadcast(comm: MarimoComm):
         assert notification.model_id == comm.comm_id
 
 
-def test_comm_broadcast_echo_update(comm: MarimoComm):
-    """echo_update should still contribute to replay state."""
+def test_comm_drops_echo_update(comm: MarimoComm):
+    """echo_update (ipywidgets' ack of a client write) is never broadcast."""
     with patch(
         "marimo._plugins.ui._impl.comm.broadcast_notification"
     ) as mock_broadcast:
         comm._broadcast(
             {"method": "echo_update", "state": {"key": "value"}},
             [],
         )
+        mock_broadcast.assert_not_called()
+
+
+def test_client_update_is_not_echoed_back(
+    comm_manager: MarimoCommManager, comm: MarimoComm
+):
+    """A client write must not bounce back as a live update.
+
+    Simulates ipywidgets' `Widget.set_state`, which acks a client
+    update with `echo_update` before observers send genuine updates.
+    Guarantees an echo is sent, unlike the real-stack test in
+    test_anywidget.py, where echoing is up to ipywidgets.
+    """
+
+    def fake_set_state(msg: dict) -> None:
+        state = msg["content"]["data"]["state"]
+        # ipywidgets acks the client's own values first...
+        comm.send({"method": "echo_update", "state": state})
+        # ...then an observer reacts with a kernel-driven change.
+        comm.send({"method": "update", "state": {"chart": "<new svg>"}})
+
+    comm.on_msg(fake_set_state)
+    command = ModelCommand(
+        model_id=comm.comm_id,
+        message=ModelUpdateMessage(state={"slider": 5}, buffer_paths=[]),
+        buffers=[],
+    )
+    with patch(
+        "marimo._plugins.ui._impl.comm.broadcast_notification"
+    ) as mock_broadcast:
+        comm_manager.receive_comm_message(command)
+        # Exactly one broadcast: the observer's update. The echo of
+        # {"slider": 5} is dropped.
         mock_broadcast.assert_called_once()
         notification = mock_broadcast.call_args[0][0]
-        assert notification.model_id == comm.comm_id
-        assert notification.message.state == {"key": "value"}
+        assert notification.message.state == {"chart": "<new svg>"}
 
 
 def test_comm_manager_receive_update_message(

tests/_session/state/test_session_view.py

@@ -45,6 +45,9 @@
     CreateNotebookCommand,
     ExecuteCellCommand,
     ExecuteCellsCommand,
+    ModelCommand,
+    ModelCustomMessage,
+    ModelUpdateMessage,
     UpdateUIElementCommand,
 )
 from marimo._session.state.session_view import ModelReplayState, SessionView
@@ -351,6 +354,111 @@ def test_model_multiple_models(session_view: SessionView) -> None:
     assert session_view.model_states[model_id2].state == {"key": "v2"}
 
 
+def _open_model(
+    session_view: SessionView,
+    model_id: WidgetModelId,
+    state: dict[str, Any],
+) -> None:
+    session_view.add_notification(
+        ModelLifecycleNotification(
+            model_id=model_id,
+            message=ModelOpen(state=state, buffer_paths=[], buffers=[]),
+        )
+    )
+
+
+def test_model_command_merges_into_replay(session_view: SessionView) -> None:
+    """A client's model write is recorded for reconnect replay."""
+    model_id = WidgetModelId("test_model")
+    _open_model(session_view, model_id, {"count": 0, "label": "hi"})
+
+    session_view.add_control_request(
+        ModelCommand(
+            model_id=model_id,
+            message=ModelUpdateMessage(state={"count": 5}, buffer_paths=[]),
+            buffers=[],
+        )
+    )
+    assert session_view.model_states[model_id].state == {
+        "count": 5,
+        "label": "hi",
+    }
+
+
+def test_model_command_merges_buffers(session_view: SessionView) -> None:
+    model_id = WidgetModelId("test_model")
+    _open_model(session_view, model_id, {"img": None})
+
+    session_view.add_control_request(
+        ModelCommand(
+            model_id=model_id,
+            message=ModelUpdateMessage(
+                state={"img": None}, buffer_paths=[["img"]]
+            ),
+            buffers=[b"\x89PNG"],
+        )
+    )
+    assert session_view.model_states[model_id].buffers == {
+        ("img",): b"\x89PNG"
+    }
+
+
+def test_model_command_strips_code_and_style(
+    session_view: SessionView,
+) -> None:
+    """Replayed state reaches future viewers, so a client must not
+    be able to persist `_esm` or `_css` into it."""
+    model_id = WidgetModelId("test_model")
+    _open_model(session_view, model_id, {"count": 0})
+
+    session_view.add_control_request(
+        ModelCommand(
+            model_id=model_id,
+            message=ModelUpdateMessage(
+                state={
+                    "_esm": "alert('pwned')",
+                    "_css": "body { display: none }",
+                    "count": 2,
+                },
+                buffer_paths=[],
+            ),
+            buffers=[],
+        )
+    )
+    assert session_view.model_states[model_id].state == {"count": 2}
+
+
+def test_model_command_without_open_ignored(
+    session_view: SessionView,
+) -> None:
+    model_id = WidgetModelId("never_opened")
+    session_view.add_control_request(
+        ModelCommand(
+            model_id=model_id,
+            message=ModelUpdateMessage(state={"count": 1}, buffer_paths=[]),
+            buffers=[],
+        )
+    )
+    assert model_id not in session_view.model_states
+
+
+def test_model_command_custom_message_ignored(
+    session_view: SessionView,
+) -> None:
+    """Custom messages are ephemeral — they never mutate replay state."""
+    model_id = WidgetModelId("test_model")
+    _open_model(session_view, model_id, {"count": 0})
+
+    session_view.add_control_request(
+        ModelCommand(
+            model_id=model_id,
+            message=ModelCustomMessage(content={"foo": "bar"}),
+            buffers=[],
+        )
+    )
+    assert session_view.model_states[model_id].state == {"count": 0}
+
+
 def test_get_model_notifications(session_view: SessionView) -> None:
     # Empty initially
     assert session_view.get_model_notifications() == []