feat(context): add context_manager="auto" facade on Agent

strands-py/src/strands/agent/agent.py

@@ -17,9 +17,11 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    Literal,
     TypeVar,
     Union,
     cast,
+    get_args,
 )
 
 from opentelemetry import trace as trace_api
@@ -107,6 +109,21 @@ class _DefaultRetryStrategySentinel:
 _DEFAULT_AGENT_NAME = "Strands Agents"
 _DEFAULT_AGENT_ID = "default"
 
+ContextManagerStrategy = Literal["auto"]
+"""Supported values for the ``context_manager`` parameter."""
+
+_CONTEXT_MANAGER_MAX_RESULT_TOKENS = 1_500
+"""Benchmark-validated token threshold for offloading tool results."""
+
+_CONTEXT_MANAGER_PREVIEW_TOKENS = 750
+"""Benchmark-validated preview token count for offloaded results."""
+
+_CONTEXT_MANAGER_SUMMARY_RATIO = 0.3
+"""Benchmark-validated ratio of messages to summarize on overflow."""
+
+_CONTEXT_MANAGER_COMPRESSION_THRESHOLD = 0.85
+"""Benchmark-validated context window ratio that triggers proactive compression."""
+
 
 class Agent(AgentBase):
     """Core Agent implementation.
@@ -141,6 +158,7 @@ def __init__(
         name: str | None = None,
         description: str | None = None,
         state: AgentState | dict | None = None,
+        context_manager: ContextManagerStrategy | None = None,
         plugins: list[Plugin] | None = None,
         hooks: list[HookProvider | HookCallback] | None = None,
         session_manager: SessionManager | None = None,
@@ -194,6 +212,15 @@ def __init__(
                 Defaults to None.
             state: stateful information for the agent. Can be either an AgentState object, or a json serializable dict.
                 Defaults to an empty AgentState object.
+            context_manager: Context management strategy. When set to ``"auto"``, composes
+                a ContextOffloader plugin (max_result_tokens=1500, preview_tokens=750) with a
+                SummarizingConversationManager (summary_ratio=0.3, compression_threshold=0.85)
+                using benchmark-validated defaults. If ``conversation_manager`` is also provided,
+                the user's conversation manager is used instead. Defaults to None (no context management).
+
+                Note: The offloader uses in-memory storage that does not persist across process
+                restarts. For agents using ``session_manager``, provide an explicit
+                ``ContextOffloader`` with durable storage via the ``plugins`` parameter.
             plugins: List of Plugin instances to extend agent functionality.
                 Plugins are initialized with the agent instance after construction and can register hooks,
                 modify agent attributes, or perform other setup tasks.
@@ -249,15 +276,21 @@ def __init__(
         else:
             self.callback_handler = callback_handler
 
-        if self.model.stateful and conversation_manager is not None:
+        if self.model.stateful and (conversation_manager is not None or context_manager is not None):
             raise ValueError(
-                "conversation_manager cannot be used with a stateful model. "
+                "context_manager and conversation_manager cannot be used with a stateful model. "
                 "The model manages conversation state server-side."
             )
 
+        resolved_conversation_manager, resolved_plugins = self._resolve_context_manager(
+            context_manager, conversation_manager, plugins
+        )
+
         self.conversation_manager: ConversationManager
         if self.model.stateful:
             self.conversation_manager = NullConversationManager()
+        elif resolved_conversation_manager:
+            self.conversation_manager = resolved_conversation_manager
         elif conversation_manager:
             self.conversation_manager = conversation_manager
         else:
@@ -378,12 +411,72 @@ def __init__(
         # Register built-in plugins
         self._plugin_registry.add_and_init(_ModelPlugin())
 
-        if plugins:
-            for plugin in plugins:
+        plugins_to_register = resolved_plugins if resolved_plugins is not None else plugins
+        if plugins_to_register:
+            for plugin in plugins_to_register:
                 self._plugin_registry.add_and_init(plugin)
 
         self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))
 
+    @staticmethod
+    def _resolve_context_manager(
+        context_manager: "ContextManagerStrategy | None",
+        conversation_manager: ConversationManager | None,
+        plugins: list[Plugin] | None,
+    ) -> tuple[ConversationManager | None, list[Plugin] | None]:
+        """Resolve context_manager facade into concrete conversation_manager and plugins.
+
+        When context_manager is None, returns (None, None) and no resolution occurs.
+        When "auto", constructs a SummarizingConversationManager and ContextOffloader
+        with benchmark-validated defaults, unless the user already provided those.
+
+        Args:
+            context_manager: The facade value ("auto" or None).
+            conversation_manager: User-provided conversation manager, takes precedence if set.
+            plugins: User-provided plugin list; offloader is appended if not already present.
+
+        Returns:
+            Tuple of (resolved conversation manager, resolved plugins list).
+            Both are None when context_manager is None.
+
+        Raises:
+            ValueError: If context_manager is not a supported value.
+        """
+        if context_manager is None:
+            return None, None
+
+        supported = get_args(ContextManagerStrategy)
+        if context_manager not in supported:
+            raise ValueError(
+                f"Unsupported context_manager value: {context_manager!r}. Supported values: {supported}"
+            )
+
+        from ..vended_plugins.context_offloader import ContextOffloader, InMemoryStorage
+        from .conversation_manager import SummarizingConversationManager
+
+        resolved_plugins = list(plugins) if plugins else []
+
+        has_offloader = any(
+            isinstance(p, ContextOffloader) for p in resolved_plugins
+        )
+        if not has_offloader:
+            offloader = ContextOffloader(
+                storage=InMemoryStorage(),
+                max_result_tokens=_CONTEXT_MANAGER_MAX_RESULT_TOKENS,
+                preview_tokens=_CONTEXT_MANAGER_PREVIEW_TOKENS,
+            )
+            resolved_plugins.append(offloader)
+
+        if conversation_manager is not None:
+            resolved_conversation_manager = conversation_manager
+        else:
+            resolved_conversation_manager = SummarizingConversationManager(
+                summary_ratio=_CONTEXT_MANAGER_SUMMARY_RATIO,
+                proactive_compression={"compression_threshold": _CONTEXT_MANAGER_COMPRESSION_THRESHOLD},
+            )
+
+        return resolved_conversation_manager, resolved_plugins
+
     def cancel(self) -> None:
         """Cancel the currently running agent invocation.
 

strands-py/tests/strands/agent/test_agent_context_manager.py

@@ -0,0 +1,102 @@
+"""Tests for the context_manager parameter on Agent."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from strands import Agent, Plugin
+from strands.agent.conversation_manager import SlidingWindowConversationManager, SummarizingConversationManager
+from strands.vended_plugins.context_offloader import ContextOffloader, InMemoryStorage
+
+
+@pytest.fixture
+def mock_model():
+    model = MagicMock()
+    model.stateful = False
+    model.context_window_limit = 200_000
+    return model
+
+
+class TestContextManagerNone:
+    def test_default_preserves_sliding_window(self, mock_model):
+        agent = Agent(model=mock_model)
+        assert isinstance(agent.conversation_manager, SlidingWindowConversationManager)
+
+    def test_explicit_none_preserves_sliding_window(self, mock_model):
+        agent = Agent(model=mock_model, context_manager=None)
+        assert isinstance(agent.conversation_manager, SlidingWindowConversationManager)
+
+    def test_no_offloader_plugin_by_default(self, mock_model):
+        agent = Agent(model=mock_model)
+        assert "context_offloader" not in agent._plugin_registry._plugins
+
+
+class TestContextManagerAuto:
+    def test_uses_summarizing_conversation_manager(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert isinstance(agent.conversation_manager, SummarizingConversationManager)
+
+    def test_summary_ratio_is_benchmark_default(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert agent.conversation_manager.summary_ratio == 0.3
+
+    def test_proactive_compression_at_85_percent(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert agent.conversation_manager._compression_threshold == 0.85
+
+    def test_adds_context_offloader_plugin(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        assert "context_offloader" in agent._plugin_registry._plugins
+
+    def test_offloader_max_result_tokens(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        offloader = agent._plugin_registry._plugins["context_offloader"]
+        assert offloader._max_result_tokens == 1500
+
+    def test_offloader_preview_tokens(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        offloader = agent._plugin_registry._plugins["context_offloader"]
+        assert offloader._preview_tokens == 750
+
+    def test_offloader_uses_in_memory_storage(self, mock_model):
+        agent = Agent(model=mock_model, context_manager="auto")
+        offloader = agent._plugin_registry._plugins["context_offloader"]
+        assert isinstance(offloader._storage, InMemoryStorage)
+
+
+class TestContextManagerCoexistence:
+    def test_user_conversation_manager_is_respected(self, mock_model):
+        user_conversation_manager = SlidingWindowConversationManager(window_size=20)
+        agent = Agent(model=mock_model, context_manager="auto", conversation_manager=user_conversation_manager)
+        assert agent.conversation_manager is user_conversation_manager
+
+    def test_offloader_still_added_with_user_conversation_manager(self, mock_model):
+        user_conversation_manager = SlidingWindowConversationManager(window_size=20)
+        agent = Agent(model=mock_model, context_manager="auto", conversation_manager=user_conversation_manager)
+        assert "context_offloader" in agent._plugin_registry._plugins
+
+    def test_user_offloader_not_overridden(self, mock_model):
+        user_offloader = ContextOffloader(storage=MagicMock(), max_result_tokens=3000, preview_tokens=1000)
+        agent = Agent(model=mock_model, context_manager="auto", plugins=[user_offloader])
+        assert agent._plugin_registry._plugins["context_offloader"]._max_result_tokens == 3000
+
+    def test_user_plugins_preserved(self, mock_model):
+        class MyPlugin(Plugin):
+            name = "my_plugin"
+
+        plugin = MyPlugin()
+        agent = Agent(model=mock_model, context_manager="auto", plugins=[plugin])
+        assert "my_plugin" in agent._plugin_registry._plugins
+        assert "context_offloader" in agent._plugin_registry._plugins
+
+
+class TestContextManagerErrors:
+    def test_raises_with_stateful_model(self):
+        stateful_model = MagicMock()
+        stateful_model.stateful = True
+        with pytest.raises(ValueError, match="stateful model"):
+            Agent(model=stateful_model, context_manager="auto")
+
+    def test_raises_with_unsupported_value(self, mock_model):
+        with pytest.raises(ValueError, match="Unsupported context_manager value"):
+            Agent(model=mock_model, context_manager="manual")

strands-ts/src/agent/__tests__/agent.context-manager.test.ts

@@ -0,0 +1,121 @@
+import { describe, expect, it } from 'vitest'
+import { Agent } from '../agent.js'
+import { MockMessageModel } from '../../__fixtures__/mock-message-model.js'
+import { SlidingWindowConversationManager } from '../../conversation-manager/sliding-window-conversation-manager.js'
+import { SummarizingConversationManager } from '../../conversation-manager/summarizing-conversation-manager.js'
+import { ContextOffloader } from '../../vended-plugins/context-offloader/plugin.js'
+import { InMemoryStorage } from '../../vended-plugins/context-offloader/storage.js'
+import type { ConversationManager } from '../../conversation-manager/conversation-manager.js'
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+function internals(agent: Agent): any {
+  return agent as any
+}
+
+function getConversationManager(agent: Agent): ConversationManager {
+  return internals(agent)._conversationManager
+}
+
+function getPending(agent: Agent): any[] {
+  return internals(agent)._pluginRegistry._pending
+}
+
+describe('Agent contextManager', () => {
+  describe('when undefined (default)', () => {
+    it('uses SlidingWindowConversationManager', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const agent = new Agent({ model })
+      expect(getConversationManager(agent)).toBeInstanceOf(SlidingWindowConversationManager)
+    })
+
+    it('does not add ContextOffloader plugin', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const agent = new Agent({ model })
+      const pending = getPending(agent)
+      expect(pending.find((p: any) => p.name === 'strands:context-offloader')).toBeUndefined()
+    })
+  })
+
+  describe('when "auto"', () => {
+    it('uses SummarizingConversationManager', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const agent = new Agent({ model, contextManager: 'auto' })
+      expect(getConversationManager(agent)).toBeInstanceOf(SummarizingConversationManager)
+    })
+
+    it('sets summaryRatio to 0.3', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const agent = new Agent({ model, contextManager: 'auto' })
+      const conversationManager = getConversationManager(agent) as any
+      expect(conversationManager._summaryRatio).toBe(0.3)
+    })
+
+    it('enables proactive compression at 0.85', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const agent = new Agent({ model, contextManager: 'auto' })
+      const conversationManager = getConversationManager(agent) as any
+      expect(conversationManager._compressionThreshold).toBe(0.85)
+    })
+
+    it('adds ContextOffloader plugin with benchmark defaults', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const agent = new Agent({ model, contextManager: 'auto' })
+      const pending = getPending(agent)
+      const offloader = pending.find((p: any) => p.name === 'strands:context-offloader') as any
+      expect(offloader).toBeDefined()
+      expect(offloader._maxResultTokens).toBe(1500)
+      expect(offloader._previewTokens).toBe(750)
+      expect(offloader._storage).toBeInstanceOf(InMemoryStorage)
+    })
+  })
+
+  describe('coexistence with conversationManager', () => {
+    it('respects user-provided conversationManager', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const userCm = new SlidingWindowConversationManager({ windowSize: 20 })
+      const agent = new Agent({ model, contextManager: 'auto', conversationManager: userCm })
+      expect(getConversationManager(agent)).toBe(userCm)
+    })
+
+    it('still adds ContextOffloader when user provides conversationManager', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const userCm = new SlidingWindowConversationManager({ windowSize: 20 })
+      const agent = new Agent({ model, contextManager: 'auto', conversationManager: userCm })
+      const pending = getPending(agent)
+      expect(pending.find((p: any) => p.name === 'strands:context-offloader')).toBeDefined()
+    })
+
+    it('does not add duplicate ContextOffloader if user provides one', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      const userOffloader = new ContextOffloader({
+        storage: new InMemoryStorage(),
+        maxResultTokens: 3000,
+        previewTokens: 1000,
+      })
+      const agent = new Agent({ model, contextManager: 'auto', plugins: [userOffloader] })
+      const pending = getPending(agent)
+      const offloaders = pending.filter((p: any) => p.name === 'strands:context-offloader')
+      expect(offloaders).toHaveLength(1)
+      expect((offloaders[0] as any)._maxResultTokens).toBe(3000)
+    })
+  })
+
+  describe('stateful model', () => {
+    it('throws when used with a stateful model', () => {
+      class StatefulModel extends MockMessageModel {
+        override get stateful(): boolean {
+          return true
+        }
+      }
+      const model = new StatefulModel().addTurn({ type: 'textBlock', text: 'hi' })
+      expect(() => new Agent({ model, contextManager: 'auto' })).toThrow('stateful model')
+    })
+  })
+
+  describe('unsupported value', () => {
+    it('throws for invalid contextManager value', () => {
+      const model = new MockMessageModel().addTurn({ type: 'textBlock', text: 'hi' })
+      expect(() => new Agent({ model, contextManager: 'manual' as any })).toThrow('Unsupported contextManager value')
+    })
+  })
+})