feat: add ReadOnlySessionManager wrapper for read-only sessions (Approach B)

Author: mohanasudhan

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,73 @@
+"""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.
+
+    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 register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
+        """Register hooks with write methods pointing to this wrapper's no-ops."""
+        super().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."""
+
+    def redact_latest_message(self, redact_message: Message, agent: "Agent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips message redaction persistence."""
+
+    def sync_agent(self, agent: "Agent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips agent sync."""
+
+    def sync_multi_agent(self, source: "MultiAgentBase", **kwargs: Any) -> None:
+        """No-op: read-only mode skips multi-agent sync."""
+
+    def append_bidi_message(self, message: Message, agent: "BidiAgent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips bidi message persistence."""
+
+    def sync_bidi_agent(self, agent: "BidiAgent", **kwargs: Any) -> None:
+        """No-op: read-only mode skips bidi agent sync."""

tests/strands/session/test_read_only_session_manager.py

@@ -0,0 +1,153 @@
+"""Tests for ReadOnlySessionManager wrapper (Approach B)."""
+
+from unittest.mock import Mock
+
+import pytest
+
+from strands.agent.agent import Agent
+from strands.hooks.events import (
+    AfterInvocationEvent,
+    AfterMultiAgentInvocationEvent,
+    AfterNodeCallEvent,
+    AgentInitializedEvent,
+    MessageAddedEvent,
+    MultiAgentInitializedEvent,
+)
+from strands.hooks.registry import HookRegistry
+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):
+    """Test that initialize restores agent state from the inner session manager."""
+    inner = existing_read_only_session_manager._inner
+    session_agent = SessionAgent(
+        agent_id="test-agent",
+        state={"key": "value"},
+        conversation_manager_state={
+            "__name__": "SlidingWindowConversationManager",
+            "removed_message_count": 0,
+        },
+    )
+    inner.session_repository.create_agent("test-session", session_agent)
+
+    message = SessionMessage(
+        message={"role": "user", "content": [ContentBlock(text="Hello")]},
+        message_id=0,
+    )
+    inner.session_repository.create_message("test-session", "test-agent", message)
+
+    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.agent_id = "test-agent"
+    source = Mock()
+
+    # None of these should raise or have side effects
+    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_register_with_wrapper_methods(read_only_session_manager):
+    """Test that hooks call the wrapper's no-op methods, not the inner's write methods."""
+    registry = HookRegistry()
+    read_only_session_manager.register_hooks(registry)
+
+    # All hooks should be registered (reads + writes), but writes point to wrapper no-ops
+    assert len(registry._registered_callbacks.get(AgentInitializedEvent, [])) == 1
+    assert len(registry._registered_callbacks.get(MessageAddedEvent, [])) == 2
+    assert len(registry._registered_callbacks.get(AfterInvocationEvent, [])) == 1
+    assert len(registry._registered_callbacks.get(MultiAgentInitializedEvent, [])) == 1
+    assert len(registry._registered_callbacks.get(AfterNodeCallEvent, [])) == 1
+    assert len(registry._registered_callbacks.get(AfterMultiAgentInvocationEvent, [])) == 1
+
+
+def test_messages_not_persisted_via_hooks(read_only_session_manager):
+    """Test that messages are not persisted when hooks fire through the wrapper."""
+    inner = read_only_session_manager._inner
+
+    Agent(agent_id="test-agent", session_manager=read_only_session_manager)
+
+    messages = inner.session_repository.list_messages("test-session", "test-agent")
+    assert len(messages) == 0
+
+
+def test_direct_write_calls_are_noop(read_only_session_manager):
+    """Test that direct calls to write methods don't persist — the key advantage over Approach A."""
+    inner = read_only_session_manager._inner
+
+    agent = Agent(agent_id="test-agent", session_manager=read_only_session_manager)
+
+    # Directly calling sync_agent on the wrapper should be a no-op
+    agent.messages.append({"role": "user", "content": [{"text": "test"}]})
+    read_only_session_manager.sync_agent(agent)
+
+    # Inner repository should not have been updated
+    session_agent = inner.session_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)
+
+    inner = read_only_session_manager._inner
+    state = inner.session_repository.read_multi_agent("test-session", "test-multi-agent")
+    assert state is not None
+
+
+def test_inner_session_manager_accessible(read_only_session_manager, inner_session_manager):
+    """Test that the inner session manager is accessible for inspection."""
+    assert read_only_session_manager._inner is inner_session_manager