feat: add ReadOnlySessionManager wrapper for read-only sessions

src/strands/session/__init__.py

@@ -4,13 +4,15 @@
 """
 
 from .file_session_manager import FileSessionManager
+from .read_only_session_manager import ReadOnlySessionManager
 from .repository_session_manager import RepositorySessionManager
 from .s3_session_manager import S3SessionManager
 from .session_manager import SessionManager
 from .session_repository import SessionRepository
 
 __all__ = [
     "FileSessionManager",
+    "ReadOnlySessionManager",
     "RepositorySessionManager",
     "S3SessionManager",
     "SessionManager",

src/strands/session/read_only_session_manager.py

@@ -0,0 +1,104 @@
+"""Read-only session manager wrapper."""
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from ..hooks.registry import HookRegistry
+from ..types.content import Message
+from .session_manager import SessionManager
+
+if TYPE_CHECKING:
+    from ..agent.agent import Agent
+    from ..experimental.bidi.agent.agent import BidiAgent
+    from ..multiagent.base import MultiAgentBase
+
+logger = logging.getLogger(__name__)
+
+
+class ReadOnlySessionManager(SessionManager):
+    """A wrapper that delegates read operations to an inner session manager and no-ops all writes.
+
+    Read-only enforcement happens at the SessionManager level — all write methods are no-ops regardless
+    of whether they are called by the Agent, custom hooks, or user code.
+
+    Attribute access is forwarded to the inner session manager, so properties like ``session_id``
+    and ``bucket`` are available directly on the wrapper.
+
+    Note:
+        The wrapper protects writes because the Agent holds a reference to this wrapper instance.
+        Bypassing the wrapper by obtaining the inner session manager and passing it directly to an
+        Agent will lose read-only protection.
+
+    Usage::
+
+        from strands import Agent
+        from strands.session import ReadOnlySessionManager, S3SessionManager
+
+        inner = S3SessionManager(session_id="tenant-123", bucket="my-bucket")
+        agent = Agent(session_manager=ReadOnlySessionManager(inner))
+    """
+
+    def __init__(self, inner: SessionManager) -> None:
+        """Initialize the ReadOnlySessionManager.
+
+        Args:
+            inner: The session manager to delegate read operations to.
+        """
+        self._inner = inner
+
+    def __getattr__(self, name: str) -> Any:
+        """Forward attribute access to the inner session manager."""
+        return getattr(self._inner, name)
+
+    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
+        """Register hooks preserving the inner session manager's custom hooks.
+
+        Patches the inner's write methods to point to this wrapper's no-ops, then
+        delegates to the inner's register_hooks. This preserves any custom read-path
+        hooks (e.g., LTM retrieval) while ensuring write-path lambdas resolve to
+        no-ops at call time via Python's late binding.
+        """
+        write_methods = [
+            "append_message", "sync_agent", "redact_latest_message",
+            "sync_multi_agent", "append_bidi_message", "sync_bidi_agent",
+        ]
+        for name in write_methods:
+            setattr(self._inner, name, getattr(self, name))
+
+        self._inner.register_hooks(registry, **kwargs)
+
+    def initialize(self, agent: "Agent", **kwargs: Any) -> None:
+        """Delegate to inner session manager to restore agent state."""
+        self._inner.initialize(agent, **kwargs)
+
+    def initialize_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
+        """Delegate to inner session manager to restore multi-agent state."""
+        self._inner.initialize_multi_agent(source, **kwargs)
+
+    def initialize_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
+        """Delegate to inner session manager to restore bidi agent state."""
+        self._inner.initialize_bidi_agent(agent, **kwargs)
+
+    def append_message(self, message: Message, agent: "Agent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips message persistence."""
+        logger.debug("skipping append_message in read-only mode")
+
+    def redact_latest_message(self, redact_message: Message, agent: "Agent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips message redaction persistence."""
+        logger.debug("skipping redact_latest_message in read-only mode")
+
+    def sync_agent(self, agent: "Agent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips agent sync."""
+        logger.debug("skipping sync_agent in read-only mode")
+
+    def sync_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
+        """No-op: read-only mode skips multi-agent sync."""
+        logger.debug("skipping sync_multi_agent in read-only mode")
+
+    def append_bidi_message(self, message: Message, agent: "BidiAgent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips bidi message persistence."""
+        logger.debug("skipping append_bidi_message in read-only mode")
+
+    def sync_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips bidi agent sync."""
+        logger.debug("skipping sync_bidi_agent in read-only mode")

tests/strands/session/test_read_only_session_manager.py

@@ -0,0 +1,129 @@
+"""Tests for ReadOnlySessionManager wrapper."""
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from strands.agent.agent import Agent
+from strands.session.read_only_session_manager import ReadOnlySessionManager
+from strands.session.repository_session_manager import RepositorySessionManager
+from strands.types.content import ContentBlock
+from strands.types.session import Session, SessionAgent, SessionMessage, SessionType
+from tests.fixtures.mock_session_repository import MockedSessionRepository
+
+
+@pytest.fixture
+def mock_repository():
+    """Create a mock repository."""
+    return MockedSessionRepository()
+
+
+@pytest.fixture
+def inner_session_manager(mock_repository):
+    """Create an inner read-write session manager."""
+    return RepositorySessionManager(
+        session_id="test-session",
+        session_repository=mock_repository,
+    )
+
+
+@pytest.fixture
+def read_only_session_manager(inner_session_manager):
+    """Create a read-only wrapper around the inner session manager."""
+    return ReadOnlySessionManager(inner_session_manager)
+
+
+@pytest.fixture
+def existing_read_only_session_manager(mock_repository):
+    """Create a read-only wrapper with a pre-existing session."""
+    session = Session(session_id="test-session", session_type=SessionType.AGENT)
+    mock_repository.create_session(session)
+    inner = RepositorySessionManager(
+        session_id="test-session",
+        session_repository=mock_repository,
+    )
+    return ReadOnlySessionManager(inner)
+
+
+def test_initialize_delegates_to_inner(existing_read_only_session_manager, mock_repository):
+    """Test that initialize restores agent state from the inner session manager."""
+    session_agent = SessionAgent(
+        agent_id="test-agent",
+        state={"key": "value"},
+        conversation_manager_state={
+            "__name__": "SlidingWindowConversationManager",
+            "removed_message_count": 0,
+        },
+    )
+    mock_repository.create_agent("test-session", session_agent)
+    mock_repository.create_message(
+        "test-session",
+        "test-agent",
+        SessionMessage(message={"role": "user", "content": [ContentBlock(text="Hello")]}, message_id=0),
+    )
+
+    agent = Agent(agent_id="test-agent")
+    existing_read_only_session_manager.initialize(agent)
+
+    assert agent.state.get("key") == "value"
+    assert len(agent.messages) == 1
+    assert agent.messages[0]["content"][0]["text"] == "Hello"
+
+
+def test_write_methods_are_noop(read_only_session_manager):
+    """Test that all write methods are no-ops and don't raise."""
+    agent = Mock(agent_id="test-agent")
+    source = Mock()
+
+    read_only_session_manager.append_message({"role": "user", "content": []}, agent)
+    read_only_session_manager.redact_latest_message({"role": "user", "content": []}, agent)
+    read_only_session_manager.sync_agent(agent)
+    read_only_session_manager.sync_multi_agent(source)
+    read_only_session_manager.append_bidi_message({"role": "user", "content": []}, agent)
+    read_only_session_manager.sync_bidi_agent(agent)
+
+
+def test_hooks_do_not_call_inner_write_methods(inner_session_manager):
+    """Test that hooks fire the wrapper's no-ops, not the inner's write methods."""
+    with (
+        patch.object(inner_session_manager, "append_message") as mock_append,
+        patch.object(inner_session_manager, "sync_agent") as mock_sync,
+    ):
+        ro = ReadOnlySessionManager(inner_session_manager)
+        Agent(agent_id="test-agent", session_manager=ro)
+
+        mock_append.assert_not_called()
+        mock_sync.assert_not_called()
+
+
+def test_messages_not_persisted_via_hooks(read_only_session_manager, mock_repository):
+    """Test that messages are not persisted when hooks fire through the wrapper."""
+    Agent(agent_id="test-agent", session_manager=read_only_session_manager)
+
+    messages = mock_repository.list_messages("test-session", "test-agent")
+    assert len(messages) == 0
+
+
+def test_direct_write_calls_are_noop(read_only_session_manager, mock_repository):
+    """Test that direct calls to write methods don't persist."""
+    agent = Agent(agent_id="test-agent", session_manager=read_only_session_manager)
+
+    agent.messages.append({"role": "user", "content": [{"text": "test"}]})
+    read_only_session_manager.sync_agent(agent)
+
+    session_agent = mock_repository.read_agent("test-session", "test-agent")
+    assert session_agent.state == {}
+
+
+def test_multi_agent_initialize_delegates(read_only_session_manager):
+    """Test that multi-agent initialize delegates to inner."""
+    mock_multi_agent = Mock()
+    mock_multi_agent.id = "test-multi-agent"
+    mock_multi_agent.serialize_state.return_value = {"id": "test-multi-agent", "state": {}}
+
+    read_only_session_manager.initialize_multi_agent(mock_multi_agent)
+
+
+def test_getattr_forwards_to_inner(read_only_session_manager, inner_session_manager):
+    """Test that attribute access is forwarded to the inner session manager."""
+    assert read_only_session_manager.session_id == inner_session_manager.session_id