feat: add context management model setting (#3128)

Author: seratch

src/agents/model_settings.py

@@ -7,6 +7,7 @@
 from openai import Omit as _Omit
 from openai._types import Body, Query
 from openai.types.responses import ResponseIncludable
+from openai.types.responses.response_create_params import ContextManagement
 from openai.types.shared import Reasoning
 from pydantic import GetCoreSchemaHandler, TypeAdapter
 from pydantic.dataclasses import dataclass
@@ -162,6 +163,13 @@ class ModelSettings:
     retry: ModelRetrySettings | None = None
     """Opt-in runner-managed retry settings for model calls."""
 
+    context_management: list[ContextManagement] | None = None
+    """Context management entries for OpenAI Responses API requests.
+
+    For example, use ``[{"type": "compaction", "compact_threshold": 200000}]``
+    to enable server-side compaction when the rendered context crosses a token threshold.
+    """
+
     def resolve(self, override: ModelSettings | None) -> ModelSettings:
         """Produce a new ModelSettings by overlaying any non-None values from the
         override on top of this instance."""

src/agents/models/openai_responses.py

@@ -852,6 +852,7 @@ def _build_response_create_kwargs(
             "prompt_cache_retention": self._non_null_or_omit(model_settings.prompt_cache_retention),
             "reasoning": self._non_null_or_omit(model_settings.reasoning),
             "metadata": self._non_null_or_omit(model_settings.metadata),
+            "context_management": self._non_null_or_omit(model_settings.context_management),
         }
         duplicate_extra_arg_keys = sorted(set(create_kwargs).intersection(extra_args))
         if duplicate_extra_arg_keys:

tests/model_settings/test_serialization.py

@@ -75,6 +75,7 @@ def test_all_fields_serialization() -> None:
                 jitter=False,
             ),
         ),
+        context_management=[{"type": "compaction", "compact_threshold": 200000}],
     )
 
     # Verify that every single field is set to a non-None value

tests/models/test_openai_responses.py

@@ -9,6 +9,7 @@
 import pytest
 from openai import NOT_GIVEN, APIConnectionError, RateLimitError, omit
 from openai.types.responses import ResponseCompletedEvent, ResponseErrorEvent
+from openai.types.responses.response_create_params import ContextManagement
 from openai.types.shared.reasoning import Reasoning
 
 from agents import (
@@ -843,6 +844,53 @@ def test_build_response_create_kwargs_includes_extra_args_prompt_cache_key():
     assert kwargs["prompt_cache_key"] == "cache-key"
 
 
+@pytest.mark.allow_call_model_methods
+def test_build_response_create_kwargs_includes_context_management():
+    client = DummyWSClient()
+    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
+    context_management: list[ContextManagement] = [
+        {"type": "compaction", "compact_threshold": 200000}
+    ]
+
+    kwargs = model._build_response_create_kwargs(
+        system_instructions=None,
+        input="hi",
+        model_settings=ModelSettings(context_management=context_management),
+        tools=[],
+        output_schema=None,
+        handoffs=[],
+        previous_response_id=None,
+        conversation_id=None,
+        stream=False,
+        prompt=None,
+    )
+
+    assert kwargs["context_management"] == context_management
+
+
+@pytest.mark.allow_call_model_methods
+def test_build_response_create_kwargs_rejects_duplicate_context_management_extra_args():
+    client = DummyWSClient()
+    model = OpenAIResponsesModel(model="gpt-4", openai_client=client)  # type: ignore[arg-type]
+
+    with pytest.raises(TypeError, match="multiple values.*context_management"):
+        model._build_response_create_kwargs(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(
+                context_management=[{"type": "compaction", "compact_threshold": 200000}],
+                extra_args={"context_management": [{"type": "compaction"}]},
+            ),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            previous_response_id=None,
+            conversation_id=None,
+            stream=False,
+            prompt=None,
+        )
+
+
 @pytest.mark.allow_call_model_methods
 @pytest.mark.asyncio
 async def test_custom_base_url_prompt_cache_key_uses_model_settings_only() -> None:

tests/test_prompt_cache_key.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import pytest
+from openai.types.responses.response_create_params import ContextManagement
 
 from agents import Agent, ModelSettings, RunConfig, Runner
 
@@ -125,7 +126,7 @@ async def test_runner_respects_existing_extra_body_prompt_cache_key() -> None:
 async def test_runner_generates_prompt_cache_key_with_unrelated_extra_args() -> None:
     model = PromptCacheFakeModel()
     model.set_next_output([get_text_message("done")])
-    model_settings = ModelSettings(extra_args={"context_management": [{"type": "compaction"}]})
+    model_settings = ModelSettings(extra_args={"service_tier": "flex"})
     agent = Agent(
         name="test",
         model=model,
@@ -137,10 +138,34 @@ async def test_runner_generates_prompt_cache_key_with_unrelated_extra_args() ->
     assert _sent_prompt_cache_key(model) is not None
     sent_model_settings = _sent_model_settings(model)
     assert sent_model_settings.extra_args == {
-        "context_management": [{"type": "compaction"}],
+        "service_tier": "flex",
         "prompt_cache_key": _sent_prompt_cache_key(model),
     }
-    assert model_settings.extra_args == {"context_management": [{"type": "compaction"}]}
+    assert model_settings.extra_args == {"service_tier": "flex"}
+
+
+@pytest.mark.asyncio
+async def test_runner_preserves_context_management_when_adding_prompt_cache_key() -> None:
+    model = PromptCacheFakeModel()
+    model.set_next_output([get_text_message("done")])
+    context_management: list[ContextManagement] = [
+        {"type": "compaction", "compact_threshold": 200000}
+    ]
+    model_settings = ModelSettings(context_management=context_management)
+    agent = Agent(
+        name="test",
+        model=model,
+        model_settings=model_settings,
+    )
+
+    await Runner.run(agent, "hi")
+
+    assert _sent_prompt_cache_key(model) is not None
+    sent_model_settings = _sent_model_settings(model)
+    assert sent_model_settings.context_management == context_management
+    assert sent_model_settings.extra_args == {"prompt_cache_key": _sent_prompt_cache_key(model)}
+    assert model_settings.context_management == context_management
+    assert model_settings.extra_args is None
 
 
 @pytest.mark.asyncio

tests/test_source_compat_constructors.py

@@ -9,6 +9,8 @@
     FunctionTool,
     HandoffInputData,
     ItemHelpers,
+    ModelRetrySettings,
+    ModelSettings,
     MultiProvider,
     RunConfig,
     RunContextWrapper,
@@ -92,6 +94,36 @@ def test_run_config_reasoning_item_id_policy_positional_binding() -> None:
     assert config.reasoning_item_id_policy == "omit"
 
 
+def test_model_settings_context_management_append_preserves_retry_position() -> None:
+    retry = ModelRetrySettings(max_retries=1)
+    settings = ModelSettings(
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+        retry,
+    )
+
+    assert settings.retry is retry
+    assert settings.context_management is None
+
+
 def test_function_tool_positional_arguments_keep_guardrail_positions() -> None:
     async def invoke(_ctx: ToolContext[Any], _args: str) -> str:
         return "ok"