feat(session): Add allow_reconnect() method to control client behavior on disconnect

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to Shiny for Python will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [UNRELEASED]
+
+### New features
+
+* Added `session.allow_reconnect()` method to enable or disable client reconnection behavior after disconnection, similar to Shiny for R's `session$allowReconnect()`.
+
 ## [1.5.0] - 2025-09-11
 
 ### New features

shiny/api-examples/session_allow_reconnect/app-core.py

@@ -0,0 +1,50 @@
+from shiny import App, Inputs, Outputs, Session, reactive, render, ui
+
+app_ui = ui.page_fluid(
+    ui.h2("Session Reconnection Control"),
+    ui.p(
+        """
+        This example demonstrates the session.allow_reconnect() method. 
+        Click a button to set the reconnection behavior, then you can test it 
+        by simulating a disconnect (e.g., close the browser tab and reopen it, 
+        or use browser developer tools to close the WebSocket connection).
+        """
+    ),
+    ui.layout_columns(
+        ui.input_action_button("allow_true", "Allow Reconnect"),
+        ui.input_action_button("allow_false", "Disallow Reconnect"),
+        ui.input_action_button("allow_force", "Force Reconnect"),
+        col_widths=[4, 4, 4],
+    ),
+    ui.output_text("status"),
+)
+
+
+def server(input: Inputs, output: Outputs, session: Session):
+    status = reactive.Value("Reconnection not configured yet")
+
+    @reactive.effect
+    @reactive.event(input.allow_true)
+    def _():
+        session.allow_reconnect(True)
+        status.set("✓ Reconnection allowed (True)")
+
+    @reactive.effect
+    @reactive.event(input.allow_false)
+    def _():
+        session.allow_reconnect(False)
+        status.set("✗ Reconnection disallowed (False)")
+
+    @reactive.effect
+    @reactive.event(input.allow_force)
+    def _():
+        session.allow_reconnect("force")
+        status.set("⚡ Reconnection forced")
+
+    @output
+    @render.text
+    def status():
+        return status.get()
+
+
+app = App(app_ui, server)

shiny/api-examples/session_allow_reconnect/app-express.py

@@ -0,0 +1,45 @@
+from shiny import reactive, render
+from shiny.express import input, session, ui
+
+ui.h2("Session Reconnection Control")
+ui.p(
+    """
+    This example demonstrates the session.allow_reconnect() method. 
+    Click a button to set the reconnection behavior, then you can test it 
+    by simulating a disconnect (e.g., close the browser tab and reopen it, 
+    or use browser developer tools to close the WebSocket connection).
+    """
+)
+
+with ui.layout_columns(col_widths=[4, 4, 4]):
+    ui.input_action_button("allow_true", "Allow Reconnect")
+    ui.input_action_button("allow_false", "Disallow Reconnect")
+    ui.input_action_button("allow_force", "Force Reconnect")
+
+status = reactive.Value("Reconnection not configured yet")
+
+
+@reactive.effect
+@reactive.event(input.allow_true)
+def _():
+    session.allow_reconnect(True)
+    status.set("✓ Reconnection allowed (True)")
+
+
+@reactive.effect
+@reactive.event(input.allow_false)
+def _():
+    session.allow_reconnect(False)
+    status.set("✗ Reconnection disallowed (False)")
+
+
+@reactive.effect
+@reactive.event(input.allow_force)
+def _():
+    session.allow_reconnect("force")
+    status.set("⚡ Reconnection forced")
+
+
+@render.text
+def status_output():
+    return status.get()

shiny/session/_session.py

@@ -492,6 +492,39 @@ def _increment_busy_count(self) -> None: ...
     @abstractmethod
     def _decrement_busy_count(self) -> None: ...
 
+    @add_example("session_allow_reconnect")
+    @abstractmethod
+    def allow_reconnect(self, value: Literal[True, False, "force"]) -> None:
+        """
+        Allow or disallow reconnection of the session.
+
+        If `value` is `True` and the app is run in a hosting environment (such as
+        Posit Connect or Shiny Server) with reconnections enabled, then when the
+        session ends due to the network connection closing, the client will attempt
+        to reconnect to the server. If a reconnection is successful, the browser
+        will send all the current input values to the new session on the server, and
+        the server will recalculate any outputs and send them back to the client.
+
+        If `value` is `False`, reconnections will be disabled (this is the default
+        state).
+
+        If `value` is `"force"`, then the client browser will always attempt to
+        reconnect. The only reason to use `"force"` is for testing on a local
+        connection (without Shiny Server or Connect).
+
+        Parameters
+        ----------
+        value
+            One of the following:
+
+            - `True`: Allow the client to reconnect to the session after a
+              disconnection (only if running in a hosting environment with
+              reconnections enabled).
+            - `False`: Do not allow the client to reconnect to the session (default).
+            - `"force"`: Force the client to always attempt to reconnect, even on
+              local connections. This is primarily useful for testing purposes.
+        """
+
 
 # ======================================================================================
 # AppSession
@@ -1085,6 +1118,11 @@ def _decrement_busy_count(self) -> None:
         if self._busy_count == 0:
             self._send_message_sync({"busy": "idle"})
 
+    def allow_reconnect(self, value: Literal[True, False, "force"]) -> None:
+        if value not in (True, False, "force"):
+            raise ValueError('value must be True, False, or "force"')
+        self._send_message_sync({"allowReconnect": value})
+
     # ==========================================================================
     # On session ended
     # ==========================================================================
@@ -1273,6 +1311,9 @@ def _increment_busy_count(self) -> None:
     def _decrement_busy_count(self) -> None:
         self._root_session._decrement_busy_count()
 
+    def allow_reconnect(self, value: Literal[True, False, "force"]) -> None:
+        self._root_session.allow_reconnect(value)
+
     def set_message_handler(
         self,
         name: str,

tests/pytest/test_session_allow_reconnect.py

@@ -0,0 +1,157 @@
+"""Tests for Session.allow_reconnect() method."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+from shiny import App, Inputs, Outputs, Session, ui
+from shiny._connection import MockConnection
+
+
+@pytest.mark.asyncio
+async def test_allow_reconnect_true():
+    """Test that allow_reconnect(True) sends the correct message."""
+    messages: list[str] = []
+
+    def server(input: Inputs, output: Outputs, session: Session):
+        session.allow_reconnect(True)
+
+    conn = MockConnection()
+    # Capture all messages sent by the session
+    original_send = conn.send
+
+    async def capture_send(message: str):
+        messages.append(message)
+        await original_send(message)
+
+    conn.send = capture_send
+
+    sess = App(ui.TagList(), server)._create_session(conn)
+
+    async def mock_client():
+        conn.cause_receive('{"method":"init","data":{}}')
+        await asyncio.sleep(0.1)  # Give time for the server to process
+        conn.cause_disconnect()
+
+    await asyncio.gather(mock_client(), sess._run())
+
+    # Check that allowReconnect message was sent
+    reconnect_messages = [
+        msg for msg in messages if "allowReconnect" in msg and msg != "{}"
+    ]
+    assert len(reconnect_messages) > 0, "No allowReconnect message found"
+
+    # Verify the message content
+    for msg in reconnect_messages:
+        parsed = json.loads(msg)
+        if "allowReconnect" in parsed:
+            assert parsed["allowReconnect"] is True
+            break
+    else:
+        pytest.fail("allowReconnect message not found in sent messages")
+
+
+@pytest.mark.asyncio
+async def test_allow_reconnect_false():
+    """Test that allow_reconnect(False) sends the correct message."""
+    messages: list[str] = []
+
+    def server(input: Inputs, output: Outputs, session: Session):
+        session.allow_reconnect(False)
+
+    conn = MockConnection()
+    original_send = conn.send
+
+    async def capture_send(message: str):
+        messages.append(message)
+        await original_send(message)
+
+    conn.send = capture_send
+
+    sess = App(ui.TagList(), server)._create_session(conn)
+
+    async def mock_client():
+        conn.cause_receive('{"method":"init","data":{}}')
+        await asyncio.sleep(0.1)
+        conn.cause_disconnect()
+
+    await asyncio.gather(mock_client(), sess._run())
+
+    # Check that allowReconnect message was sent with False
+    reconnect_messages = [
+        msg for msg in messages if "allowReconnect" in msg and msg != "{}"
+    ]
+    assert len(reconnect_messages) > 0, "No allowReconnect message found"
+
+    for msg in reconnect_messages:
+        parsed = json.loads(msg)
+        if "allowReconnect" in parsed:
+            assert parsed["allowReconnect"] is False
+            break
+    else:
+        pytest.fail("allowReconnect message not found in sent messages")
+
+
+@pytest.mark.asyncio
+async def test_allow_reconnect_force():
+    """Test that allow_reconnect('force') sends the correct message."""
+    messages: list[str] = []
+
+    def server(input: Inputs, output: Outputs, session: Session):
+        session.allow_reconnect("force")
+
+    conn = MockConnection()
+    original_send = conn.send
+
+    async def capture_send(message: str):
+        messages.append(message)
+        await original_send(message)
+
+    conn.send = capture_send
+
+    sess = App(ui.TagList(), server)._create_session(conn)
+
+    async def mock_client():
+        conn.cause_receive('{"method":"init","data":{}}')
+        await asyncio.sleep(0.1)
+        conn.cause_disconnect()
+
+    await asyncio.gather(mock_client(), sess._run())
+
+    # Check that allowReconnect message was sent with "force"
+    reconnect_messages = [
+        msg for msg in messages if "allowReconnect" in msg and msg != "{}"
+    ]
+    assert len(reconnect_messages) > 0, "No allowReconnect message found"
+
+    for msg in reconnect_messages:
+        parsed = json.loads(msg)
+        if "allowReconnect" in parsed:
+            assert parsed["allowReconnect"] == "force"
+            break
+    else:
+        pytest.fail("allowReconnect message not found in sent messages")
+
+
+def test_allow_reconnect_invalid_value():
+    """Test that allow_reconnect raises ValueError for invalid values."""
+    from shiny._connection import MockConnection
+    from shiny.session._session import AppSession
+
+    conn = MockConnection()
+    sess = AppSession(
+        app=App(ui.TagList(), lambda i, o, s: None), id="test", conn=conn
+    )
+
+    # Test invalid values
+    with pytest.raises(ValueError, match='value must be True, False, or "force"'):
+        sess.allow_reconnect("invalid")  # type: ignore
+
+    with pytest.raises(ValueError, match='value must be True, False, or "force"'):
+        sess.allow_reconnect(None)  # type: ignore
+
+    with pytest.raises(ValueError, match='value must be True, False, or "force"'):
+        sess.allow_reconnect(1)  # type: ignore