feat(router): native compact-to-fit across conversation APIs (#42074)

* feat(router): native compact-to-fit across conversation APIs

* fix(router): preserve compaction admission and shared client boundaries

* fix(router): honor compaction fit fallbacks and router-scoped access

* fix(router): charge compaction usage to caller token limits

* test(http): keep FastAPI inside proxy tests

* fix(router): check compactor capacity before skipping escalation

litellm/anthropic_beta_headers_config.json

@@ -7,6 +7,7 @@
     "bash_20250124": null,
     "code-execution-2025-08-25": "code-execution-2025-08-25",
     "compact-2026-01-12": "compact-2026-01-12",
+    "compact-2026-09-04": "compact-2026-09-04",
     "computer-use-2025-01-24": "computer-use-2025-01-24",
     "computer-use-2025-11-24": "computer-use-2025-11-24",
     "context-1m-2025-08-07": "context-1m-2025-08-07",

litellm/litellm_core_utils/initialize_dynamic_callback_params.py

@@ -1,10 +1,24 @@
 import re
-from collections.abc import Iterator, Mapping
+from collections.abc import Generator, Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
 from typing import Any, Final
 
 from litellm.types.utils import OTEL_SPAN_SCOPES, TRUSTED_CALLBACK_VARS_FIELD, StandardCallbackDynamicParams
 
 _CLIENT_CALLBACK_METADATA_SLOTS: Final[tuple[str, ...]] = ("litellm_metadata", "metadata")
+_inherited_message_logging_disabled: Final[ContextVar[bool]] = ContextVar(
+    "inherited_message_logging_disabled", default=False
+)
+
+
+@contextmanager
+def inherit_message_logging_privacy(disabled: bool) -> Generator[None]:
+    token: Final = _inherited_message_logging_disabled.set(_inherited_message_logging_disabled.get() or disabled)
+    try:
+        yield
+    finally:
+        _inherited_message_logging_disabled.reset(token)
 
 
 def iter_client_callback_metadata_dicts(
@@ -143,7 +157,7 @@ def get_trusted_callback_params(kwargs: Mapping[str, Any] | None) -> tuple[tuple
 
 
 def initialize_standard_callback_dynamic_params(
-    kwargs: dict | None = None,
+    kwargs: dict[str, object] | None = None,
 ) -> StandardCallbackDynamicParams:
     """
     Initialize the standard callback dynamic params from the kwargs
@@ -179,4 +193,10 @@ def initialize_standard_callback_dynamic_params(
             if param in _trusted_overlay_callback_params:
                 standard_callback_dynamic_params[param] = trusted_value
 
+    if _inherited_message_logging_disabled.get():
+        private_params: Final[StandardCallbackDynamicParams] = {
+            **standard_callback_dynamic_params,
+            "turn_off_message_logging": True,
+        }
+        return private_params
     return standard_callback_dynamic_params

litellm/llms/anthropic/chat/transformation.py

@@ -96,6 +96,7 @@ from ..common_utils import (
     AnthropicModelInfo,
     eager_input_streaming_flag,
     process_anthropic_headers,
+    requires_native_compaction_beta,
     strip_advisor_blocks_from_messages,
 )
 
@@ -1770,7 +1771,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
             )
         return tools
 
-    def _ensure_beta_header(self, headers: dict, beta_value: str) -> None:
+    def _ensure_beta_header(self, headers: dict[str, str], beta_value: str) -> None:
         """
         Ensure a beta header value is present in the anthropic-beta header.
         Merges with existing values instead of overriding them.
@@ -1779,13 +1780,17 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
             headers: Dictionary of headers to update
             beta_value: The beta header value to add
         """
-        existing_beta: Final = headers.get("anthropic-beta")
-        if existing_beta is None:
-            headers["anthropic-beta"] = beta_value
-            return
-        existing_values: Final = [beta.strip() for beta in existing_beta.split(",")]
-        if beta_value not in existing_values:
-            headers["anthropic-beta"] = f"{existing_beta}, {beta_value}"
+        existing_values: Final = tuple(
+            beta.strip()
+            for key, value in headers.items()
+            if key.lower() == "anthropic-beta"
+            for beta in value.split(",")
+            if beta.strip()
+        )
+        for key in tuple(headers):
+            if key.lower() == "anthropic-beta":
+                headers.pop(key)
+        headers["anthropic-beta"] = ", ".join(dict.fromkeys((*existing_values, beta_value)))
 
     def _ensure_context_management_beta_header(self, headers: dict, context_management: object) -> None:
         """
@@ -1823,7 +1828,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
                 ANTHROPIC_BETA_HEADER_VALUES.CONTEXT_MANAGEMENT_2025_06_27.value,
             )
 
-    def update_headers_with_optional_anthropic_beta(self, headers: dict, optional_params: dict) -> dict:
+    def update_headers_with_optional_anthropic_beta(
+        self, headers: dict, optional_params: dict, messages: Sequence[object] = ()
+    ) -> dict:
         """Update headers with optional anthropic beta."""
 
         # Skip adding beta headers for Vertex requests
@@ -1832,6 +1839,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
         if is_vertex_request:
             return headers
 
+        if requires_native_compaction_beta(self._resolved_provider, optional_params, messages):
+            self._ensure_beta_header(headers, ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_09_04.value)
+
         _tools: Final = optional_params.get("tools", [])
         for tool in _tools:
             if tool.get("type", None) and tool.get("type").startswith(ANTHROPIC_HOSTED_TOOLS.WEB_FETCH.value):
@@ -1928,8 +1938,6 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
             custom_llm_provider=self._resolved_provider,
         )
 
-        headers = self.update_headers_with_optional_anthropic_beta(headers=headers, optional_params=optional_params)
-
         # === Tool-name sanitization (single chokepoint) ===
         # Anthropic enforces ^[a-zA-Z0-9_-]{1,128}$ on every tool name. We
         # sanitize *here* -- not in map_openai_params -- because:
@@ -1976,6 +1984,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
                 message=f"{e}\nReceived Messages={messages}",
             )  # don't use verbose_logger.exception, if exception is raised
 
+        self.update_headers_with_optional_anthropic_beta(
+            headers=headers, optional_params=optional_params, messages=anthropic_messages
+        )
+
         ## Auto-strip advisor blocks from history if advisor tool is absent.
         ## Prevents Anthropic 400: advisor_tool_result in history requires advisor tool.
         _all_tools: Final = optional_params.get("tools") or []

litellm/llms/anthropic/common_utils.py

@@ -79,6 +79,27 @@ _CLAUDE_CODE_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
 _CLAUDE_CODE_USER_AGENT_PREFIXES: Final = ("claude-cli/", "claude-code/")
 
 
+def requires_native_compaction_beta(
+    custom_llm_provider: str,
+    optional_params: Mapping[str, object],
+    messages: Sequence[object],
+) -> bool:
+    return custom_llm_provider == "anthropic" and (
+        optional_params.get("compaction") is not None
+        or any(
+            isinstance(block, Mapping)
+            and block.get("type") == "compaction"
+            and isinstance(block.get("signature"), str)
+            and bool(block.get("signature"))
+            for message in messages
+            if isinstance(message, Mapping)
+            for content in (message.get("content"),)
+            if isinstance(content, (list, tuple))
+            for block in content
+        )
+    )
+
+
 def supports_anthropic_cache_control(model: str, custom_llm_provider: str | None) -> bool:
     from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
     from litellm.utils import supports_prompt_caching

litellm/llms/anthropic/compaction.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final
+
+from pydantic import TypeAdapter
+
+from litellm.llms.compaction import CompactionProtocol
+from litellm.types.llms.anthropic import ANTHROPIC_BETA_HEADER_VALUES, AnthropicCompaction
+
+_MAPPING: Final = TypeAdapter(Mapping[str, object])
+_OBJECTS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
+_HEADERS: Final = TypeAdapter(dict[str, str])
+_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
+_CONFLICTS: Final = ("context_management", "response_format", "stop", "stop_sequences", "tool_choice")
+
+
+def supports_native_compaction(params: Mapping[str, object]) -> bool:
+    from litellm.utils import get_model_info
+
+    if params.get("custom_llm_provider") not in (None, "anthropic", "openai"):
+        return False
+    model: Final = str(params.get("model", "")).removeprefix("openai/").removeprefix("anthropic/")
+    try:
+        return get_model_info(model=model, custom_llm_provider="anthropic").get("supports_anthropic_compaction") is True
+    except Exception:
+        return False
+
+
+def compatible_defaults(payload: Mapping[str, object]) -> bool:
+    return all(payload.get(key) is None for key in _CONFLICTS)
+
+
+def request_kwargs() -> Mapping[str, object]:
+    operation: Final[AnthropicCompaction] = {"type": "summarize"}
+    return MappingProxyType(
+        {
+            "compaction": operation,
+            "extra_headers": _HEADERS.validate_python(
+                MappingProxyType({"anthropic-beta": ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_09_04.value})
+            ),
+        }
+    )
+
+
+def _native_blocks(protocol: CompactionProtocol, response: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
+    if protocol == "messages":
+        return (
+            _OBJECTS.validate_python(response.get("content", ())) if response.get("stop_reason") == "compaction" else ()
+        )
+    choices: Final = _OBJECTS.validate_python(response.get("choices", ()))
+    choice: Final = choices[0] if len(choices) == 1 else _EMPTY
+    message: Final = _MAPPING.validate_python(choice.get("message", _EMPTY))
+    fields: Final = _MAPPING.validate_python(message.get("provider_specific_fields") or _EMPTY)
+    return _OBJECTS.validate_python(fields.get("compaction_blocks", ()))
+
+
+def extract_summary(protocol: CompactionProtocol, response: Mapping[str, object]) -> str | None:
+    blocks: Final = _native_blocks(protocol, response)
+    block: Final = blocks[0] if len(blocks) == 1 else _EMPTY
+    content: Final = block.get("content")
+    return (
+        content
+        if block.get("type") == "compaction"
+        and isinstance(block.get("signature"), str)
+        and block.get("signature")
+        and isinstance(content, str)
+        and content.strip()
+        else None
+    )

litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py

@@ -2,8 +2,11 @@ import copy
 import hashlib
 import json
 from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, TypeVar, cast
 
+from pydantic import JsonValue, TypeAdapter
+
 import litellm
 from litellm.llms.anthropic.experimental_pass_through.utils import (
     is_reasoning_auto_summary_enabled,
@@ -16,6 +19,7 @@ OPENAI_MAX_TOOL_NAME_LENGTH: Final = 64
 TOOL_NAME_HASH_LENGTH: Final = 8
 TOOL_NAME_PREFIX_LENGTH: Final = OPENAI_MAX_TOOL_NAME_LENGTH - TOOL_NAME_HASH_LENGTH - 1  # 55
 PROVIDERS_PROXYING_AN_UNKNOWN_BACKEND: Final = frozenset({"litellm_proxy"})
+_COMPACTION_BLOCK: Final = TypeAdapter(dict[str, JsonValue])
 
 
 def _optional_attr(source: object, name: str) -> object:
@@ -36,6 +40,20 @@ def _thought_signature(provider_specific_fields: object) -> str | None:
     return signature if isinstance(signature, str) else None
 
 
+def _compaction_blocks(provider_specific_fields: object) -> tuple[Mapping[str, object], ...]:
+    fields: Final = _as_string_mapping(provider_specific_fields)
+    raw_blocks: Final = fields.get("compaction_blocks") if fields is not None else None
+    return (
+        tuple(
+            block
+            for raw_block in raw_blocks
+            if (block := _as_string_mapping(raw_block)) is not None and block.get("type") == "compaction"
+        )
+        if isinstance(raw_blocks, (list, tuple))
+        else ()
+    )
+
+
 _ANTHROPIC_TOOL_SCHEMA_KEYS: Final = frozenset(
     {"name", "type", "input_schema", "description", "cache_control", "strict"}
 )
@@ -1330,7 +1348,11 @@ class LiteLLMAnthropicMessagesAdapter:
         tool_name_mapping: dict[str, str] | None = None,
     ) -> list[dict[str, Any]]:
         new_content: Final[list[dict[str, Any]]] = []
-        for choice in choices:
+        for choice, compaction_blocks in (
+            (choice, _compaction_blocks(_optional_attr(choice.message, "provider_specific_fields")))
+            for choice in choices
+        ):
+            new_content.extend(_COMPACTION_BLOCK.validate_python(block) for block in compaction_blocks)
             # Handle thinking blocks first
             if hasattr(choice.message, "thinking_blocks") and choice.message.thinking_blocks:
                 for thinking_block in choice.message.thinking_blocks:
@@ -1365,7 +1387,7 @@ class LiteLLMAnthropicMessagesAdapter:
                 )
 
             # Handle text content
-            if choice.message.content is not None:
+            if choice.message.content is not None and (choice.message.content != "" or not compaction_blocks):
                 new_content.append(
                     AnthropicResponseContentBlockText(type="text", text=choice.message.content).model_dump()
                 )
@@ -1545,21 +1567,35 @@ class LiteLLMAnthropicMessagesAdapter:
             openai_finish_reason=openai_finish_reason
         )
         anthropic_finish_reason: Final = (
-            "refusal"
+            "compaction"
+            if len(anthropic_content) == 1 and anthropic_content[0].get("type") == "compaction"
+            else "refusal"
             if refusal_text is not None and translated_finish_reason != "max_tokens"
             else translated_finish_reason
         )
         # extract usage
         usage: Final[Usage] = getattr(response, "usage")
-        anthropic_usage: Final = self._translate_openai_usage_to_anthropic_usage(usage)
-
-        if polyfill_result is not None and polyfill_result.iterations_usage is not None:
-            message_iteration: Final[UsageIteration] = {
-                "type": "message",
-                "input_tokens": anthropic_usage["input_tokens"],
-                "output_tokens": usage.completion_tokens or 0,
-            }
-            anthropic_usage["iterations"] = list(polyfill_result.iterations_usage) + [message_iteration]
+        message_usage: Final = self._translate_openai_usage_to_anthropic_usage(usage)
+        polyfill_iterations: Final = polyfill_result.iterations_usage if polyfill_result is not None else None
+        anthropic_usage: Final[AnthropicUsage] = (
+            TypeAdapter(AnthropicUsage).validate_python(
+                MappingProxyType(
+                    {
+                        **message_usage,
+                        "iterations": (
+                            *polyfill_iterations,
+                            UsageIteration(
+                                type="message",
+                                input_tokens=message_usage.get("input_tokens", 0),
+                                output_tokens=usage.completion_tokens or 0,
+                            ),
+                        ),
+                    }
+                )
+            )
+            if polyfill_iterations is not None
+            else message_usage
+        )
 
         translated_obj: Final = AnthropicMessagesResponse(
             id=response.id,

litellm/llms/anthropic/experimental_pass_through/messages/handler.py

@@ -580,7 +580,9 @@ def anthropic_messages_handler(
         )
     if anthropic_messages_provider_config is None:
         # Route to Responses API for OpenAI / Azure, chat/completions for everything else.
-        if _should_route_to_responses_api(custom_llm_provider, original_model, model):
+        if kwargs.get("compaction") is None and _should_route_to_responses_api(
+            custom_llm_provider, original_model, model
+        ):
             return LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
                 max_tokens=max_tokens,
                 messages=messages,

litellm/llms/anthropic/experimental_pass_through/messages/transformation.py

@@ -24,6 +24,7 @@ from ...common_utils import (
     AnthropicError,
     AnthropicModelInfo,
     optionally_handle_anthropic_oauth,
+    requires_native_compaction_beta,
     strip_advisor_blocks_from_messages,
     strip_encrypted_reasoning_blocks_from_anthropic_messages,
 )
@@ -74,6 +75,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
             "tool_choice",
             "thinking",
             "context_management",
+            *(("compaction",) if self._resolved_provider == "anthropic" else ()),
             "output_format",
             "inference_geo",
             "speed",
@@ -637,6 +639,9 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
         )
         beta_values.update(existing_beta)
 
+        if requires_native_compaction_beta(custom_llm_provider, optional_params, messages):
+            beta_values.add(ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_09_04.value)
+
         # Check for context management
         context_management_param: Final = optional_params.get("context_management")
         if context_management_param is not None:

litellm/llms/compaction.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, Literal, Protocol, TypeAlias
+
+from pydantic import TypeAdapter
+
+from litellm.types.llms.openai import AllMessageValues
+
+if TYPE_CHECKING:
+    from litellm.router import Router
+
+CompactionProtocol: TypeAlias = Literal["chat", "messages"]
+_MAPPING: Final = TypeAdapter(Mapping[str, object])
+_MESSAGES: Final = TypeAdapter(list[AllMessageValues])
+
+
+class NativeCompactionProvider(Protocol):
+    def supports_native_compaction(self, params: Mapping[str, object]) -> bool: ...
+
+    def compatible_defaults(self, payload: Mapping[str, object]) -> bool: ...
+
+    def request_kwargs(self) -> Mapping[str, object]: ...
+
+    def extract_summary(self, protocol: CompactionProtocol, response: Mapping[str, object]) -> str | None: ...
+
+
+def get_native_compaction_provider(params: Mapping[str, object]) -> NativeCompactionProvider | None:
+    from litellm.llms.anthropic import compaction
+
+    return compaction if compaction.supports_native_compaction(params) else None
+
+
+async def dispatch(router: Router, protocol: CompactionProtocol, payload: Mapping[str, object]) -> Mapping[str, object]:
+    if protocol == "messages":
+        return _MAPPING.validate_python(
+            await router.aanthropic_messages(custom_llm_provider=None, client=None, **payload)
+        )
+    response: Final = await router.acompletion(
+        model=str(payload["model"]),
+        messages=_MESSAGES.validate_python(payload["messages"]),
+        stream=False,
+        **MappingProxyType(
+            {key: value for key, value in payload.items() if key not in ("model", "messages", "stream")}
+        ),
+    )
+    return _MAPPING.validate_python(response.model_dump())

litellm/llms/custom_httpx/asgi_handler.py

@@ -0,0 +1,46 @@
+from collections.abc import Generator
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from typing import Final
+
+import httpx
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from litellm.llms.custom_httpx.http_handler import (
+    get_async_httpx_client,  # pyright: ignore[reportUnknownVariableType]  # shared cache retains its legacy parameter mapping
+)
+from litellm.types.llms.custom_http import httpxSpecialProvider
+
+
+@dataclass(frozen=True, slots=True)
+class _ASGITarget:
+    app: ASGIApp
+    root_path: str
+    client: tuple[str, int] | None
+
+
+_target: Final[ContextVar[_ASGITarget]] = ContextVar("httpx_asgi_target")
+
+
+async def _dispatch(scope: Scope, receive: Receive, send: Send) -> None:
+    target: Final = _target.get()
+    await target.app({**scope, "root_path": target.root_path, "client": target.client}, receive, send)
+
+
+_TRANSPORT: Final = httpx.ASGITransport(app=_dispatch, raise_app_exceptions=False)
+
+
+@contextmanager
+def get_async_asgi_client(
+    app: ASGIApp, root_path: str = "", client: tuple[str, int] | None = None
+) -> Generator[httpx.AsyncClient]:
+    handler: Final = get_async_httpx_client(
+        llm_provider=httpxSpecialProvider.ASGI,
+        params={"transport": _TRANSPORT, "timeout": httpx.Timeout(None), "follow_redirects": False},
+    )
+    token: Final = _target.set(_ASGITarget(app, root_path, client))
+    try:
+        yield handler.client
+    finally:
+        _target.reset(token)

litellm/llms/custom_httpx/http_handler.py

@@ -614,11 +614,15 @@ class AsyncHTTPHandler:
         client_alias: str | None = None,  # name for client in logs
         ssl_verify: VerifyTypes | None = None,
         shared_session: Optional["ClientSession"] = None,
+        transport: httpx.AsyncBaseTransport | None = None,
+        follow_redirects: bool = True,
     ):
         self.timeout = timeout
         self.event_hooks = event_hooks
         self.ssl_verify = ssl_verify
         self.shared_session = shared_session
+        self.transport = transport
+        self.follow_redirects = follow_redirects
         self._owns_client = True
         self._client = self.create_client(
             timeout=timeout,
@@ -651,6 +655,16 @@ class AsyncHTTPHandler:
         ssl_verify: VerifyTypes | None = None,
         shared_session: Optional["ClientSession"] = None,
     ) -> httpx.AsyncClient:
+        if self.transport is not None:
+            return httpx.AsyncClient(
+                transport=self.transport,
+                event_hooks=event_hooks,
+                timeout=timeout if timeout is not None else _DEFAULT_TIMEOUT,
+                headers=get_default_headers(),
+                cookies=blocked_cookie_jar(),
+                follow_redirects=self.follow_redirects,
+                trust_env=False,
+            )
         # Get unified SSL configuration
         ssl_config: Final = get_ssl_configuration(ssl_verify)
 
@@ -680,7 +694,7 @@ class AsyncHTTPHandler:
             cert=cert,
             headers=default_headers,
             cookies=blocked_cookie_jar(),
-            follow_redirects=True,
+            follow_redirects=self.follow_redirects,
             http2=http2_enabled(),
         )
 

litellm/model_prices_and_context_window_backup.json

@@ -14516,6 +14516,7 @@
         "source": "https://docs.anthropic.com/en/docs/about-claude/pricing"
     },
     "claude-sonnet-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 2.5e-06,
         "cache_creation_input_token_cost_above_1hr": 4e-06,
         "cache_read_input_token_cost": 2e-07,
@@ -14555,6 +14556,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-sonnet-4-6": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 3.75e-06,
         "cache_creation_input_token_cost_above_1hr": 6e-06,
         "cache_read_input_token_cost": 3e-07,
@@ -14770,6 +14772,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-6": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14808,6 +14811,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-6-20260205": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14845,6 +14849,7 @@
         "prompt_cache_min_tokens": 4096
     },
     "claude-opus-4-7": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14884,6 +14889,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-7-20260416": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14922,6 +14928,7 @@
         "prompt_cache_min_tokens": 2048
     },
     "claude-fable-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 1e-06,
@@ -14961,6 +14968,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-fable-5-1": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 2.5e-07,
@@ -15001,6 +15009,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -15043,6 +15052,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-8": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -62148,6 +62158,7 @@
         "supports_audio_output": true
     },
     "claude-mythos-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 1e-06,
@@ -62187,6 +62198,7 @@
         }
     },
     "claude-mythos-5-1": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 2.5e-07,
@@ -62227,6 +62239,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-mythos-preview": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 1e-06,

litellm/proxy/auth/auth_checks.py

@@ -4310,7 +4310,10 @@ def _can_object_call_model(
             )
         return True
 
-    potential_models: Final = [model]
+    from litellm.router_strategy.complexity_router.context_compaction import native_compaction_parent
+
+    compaction_parent: Final = native_compaction_parent(model)
+    potential_models: Final = [model, compaction_parent] if compaction_parent is not None else [model]
     if model in litellm.model_alias_map:
         potential_models.append(litellm.model_alias_map[model])
     elif llm_router and model in llm_router.model_group_alias:

litellm/proxy/common_request_processing.py

@@ -103,6 +103,7 @@ from litellm.proxy.common_utils.sse_keepalive import (
 )
 from litellm.proxy.dd_span_tagger import DDSpanTagger
 from litellm.proxy.guardrails.auto_router_compression import arm_pre_call as _arm_auto_router_compression
+from litellm.proxy.native_compaction import with_proxy_compaction_executor
 from litellm.proxy.route_llm_request import route_request
 from litellm.proxy.utils import ProxyLogging, _check_and_merge_model_level_guardrails
 from litellm.router import Router
@@ -2555,7 +2556,7 @@ class ProxyBaseLLMRequestProcessing:
             user_model=user_model,
             user_api_key_dict=user_api_key_dict,
         )
-        llm_call_task: Final = asyncio.create_task(llm_call)
+        llm_call_task: Final = asyncio.create_task(with_proxy_compaction_executor(llm_call, request))
         tasks.append(llm_call_task)
 
         llm_responses: Final = asyncio.gather(*tasks)  # run the moderation check in parallel to the actual llm api call

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -4475,9 +4475,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
         # 'metadata' and 'litellm_metadata' fields from litellm_params
         standard_logging_object: Final = kwargs.get("standard_logging_object") or {}
         request_metadata: Final = get_litellm_metadata_from_kwargs(kwargs)
-        if request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY):
-            # Internal sub-calls bill spend to the caller but are not the caller's
-            # traffic; charging them here would let background evals eat TPM headroom.
+        origin: Final = request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY)
+        if origin and origin != "autorouter_compaction":
+            # Background evaluations keep their exemption; foreground compaction
+            # is necessary caller traffic and consumes the caller's token limits.
             return []
         standard_logging_metadata: Final = standard_logging_object.get("metadata") or {}
 

litellm/proxy/native_compaction.py

@@ -0,0 +1,90 @@
+import asyncio
+from collections.abc import Awaitable, Mapping
+from contextvars import Context
+from types import MappingProxyType
+from typing import Final, Literal, TypeVar
+
+from fastapi import Request
+from pydantic import TypeAdapter, ValidationError
+from starlette.types import ASGIApp
+
+from litellm.exceptions import BadRequestError
+from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+    inherit_message_logging_privacy,
+    initialize_standard_callback_dynamic_params,
+)
+from litellm.llms.custom_httpx.asgi_handler import get_async_asgi_client
+from litellm.proxy.litellm_pre_call_utils import UNTRUSTED_REQUEST_HEADER_CONTROL_FIELDS
+from litellm.router_strategy.complexity_router.context_compaction import (
+    compaction_executor,
+    native_compaction_call,
+)
+
+_ResultT: Final = TypeVar("_ResultT")
+_ASGI_APP: Final = TypeAdapter[ASGIApp](ASGIApp)
+_ROOT_PATH: Final = TypeAdapter(str)
+_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
+_REMOVED_HEADERS: Final = frozenset(
+    (
+        b"content-length",
+        b"x-litellm-call-id",
+        b"x-litellm-num-retries",
+        b"x-litellm-timeout",
+        b"x-litellm-stream-timeout",
+    )
+)
+
+
+async def with_proxy_compaction_executor(call: Awaitable[_ResultT], request: Request) -> _ResultT:
+    async def execute(
+        protocol: Literal["chat", "messages"], payload: Mapping[str, object], parent_model: str | None = None
+    ) -> Mapping[str, object]:
+        logging_disabled: Final = initialize_standard_callback_dynamic_params().get("turn_off_message_logging") is True
+
+        async def dispatch() -> Mapping[str, object]:
+            scope: Final = _JSON_OBJECT.validate_python(request.scope)
+            root_path: Final = _ROOT_PATH.validate_python(scope.get("root_path", ""))
+            path: Final = "/v1/chat/completions" if protocol == "chat" else "/v1/messages"
+            url: Final = str(request.url.replace(path=root_path.rstrip("/") + path, query="", fragment=""))
+            headers: Final = tuple(
+                (name, value)
+                for name, value in request.headers.raw
+                if name.lower() not in _REMOVED_HEADERS
+                and not (logging_disabled and name.decode("latin-1").lower() in UNTRUSTED_REQUEST_HEADER_CONTROL_FIELDS)
+            )
+            with (
+                native_compaction_call(parent_model, str(payload["model"])),
+                inherit_message_logging_privacy(logging_disabled),
+            ):
+                with get_async_asgi_client(
+                    app=_ASGI_APP.validate_python(scope["app"]),
+                    root_path=root_path,
+                    client=request.client,
+                ) as client:
+                    async with client.stream(
+                        "POST", url, headers=headers, json=_JSON_OBJECT.validate_python(payload)
+                    ) as response:
+                        if not response.is_success:
+                            raise BadRequestError(
+                                message=f"Native compaction child request failed (HTTP {response.status_code})",
+                                model="context_compaction",
+                                llm_provider="",
+                            )
+                        body: Final = await response.aread()
+                        try:
+                            return MappingProxyType(_JSON_OBJECT.validate_json(body))
+                        except ValidationError:
+                            raise BadRequestError(
+                                message="Native compaction child returned an invalid JSON object",
+                                model="context_compaction",
+                                llm_provider="",
+                            ) from None
+
+        task: Final = Context().run(asyncio.create_task, dispatch())
+        return await task
+
+    token: Final = compaction_executor.set(execute)
+    try:
+        return await call
+    finally:
+        compaction_executor.reset(token)

litellm/router.py

@@ -118,6 +118,15 @@ from litellm.llms.openai_like.model_info import (
     get_openai_compatible_model_info,
 )
 from litellm.router_strategy.budget_limiter import RouterBudgetLimiting
+from litellm.router_strategy.complexity_router.context_compaction import (
+    arm_compaction,
+    compact_to_fit,
+    compaction_pending,
+    initialize_compaction_state,
+    is_native_compaction_call,
+    reject_recursive_compactor,
+    surface_for_call,
+)
 from litellm.router_strategy.least_busy import LeastBusyLoggingHandler
 from litellm.router_strategy.lowest_cost import LowestCostLoggingHandler
 from litellm.router_strategy.lowest_latency import LowestLatencyLoggingHandler
@@ -3635,6 +3644,7 @@ class Router:
                 kwargs=kwargs,
                 client_type="max_parallel_requests",
             )
+            compacted_input: Final = await compact_to_fit(self, deployment, input_kwargs, "chat")
             async with contextlib.AsyncExitStack() as deployment_slot:
                 if isinstance(max_parallel_requests_limit, MaxParallelRequestsLimit):
                     deployment_slot.enter_context(max_parallel_requests_limit)
@@ -3643,7 +3653,7 @@ class Router:
                     logging_obj=logging_obj,
                     parent_otel_span=parent_otel_span,
                 )
-                response = await litellm.acompletion(**input_kwargs)
+                response = await litellm.acompletion(**compacted_input)
 
                 ## CHECK CONTENT FILTER ERROR ##
                 if isinstance(response, ModelResponse):
@@ -5247,8 +5257,14 @@ class Router:
             if custom_llm_provider is not None:
                 response_kwargs["custom_llm_provider"] = custom_llm_provider
 
+            compacted_input: Final = await compact_to_fit(
+                self,
+                deployment,
+                response_kwargs,
+                surface_for_call(getattr(original_generic_function, "__name__", "")),
+            )
             async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
-                response = await original_generic_function(**response_kwargs)
+                response = await original_generic_function(**compacted_input)
 
             if self._should_raise_anthropic_refusal_error(
                 model=model,
@@ -7382,6 +7398,11 @@ class Router:
         If it fails after num_retries, fall back to another model group
         """
         model_group: Final[str | None] = kwargs.get("model")
+        compaction_surface: Final = surface_for_call(
+            getattr(kwargs.get("original_generic_function") or kwargs.get("original_function"), "__name__", "")
+        )
+        if compaction_surface is not None:
+            kwargs["_context_compaction_state"] = initialize_compaction_state(kwargs, compaction_surface)
         clear_pre_routing_selection(kwargs)  # pyright: ignore[reportUnknownArgumentType]  # **kwargs is untyped at this boundary
         if not isinstance(kwargs.get("attempted_targets"), AttemptedFallbackTargets):
             _fallback_metadata_key: Final = _get_router_metadata_variable_name(
@@ -12090,8 +12111,8 @@ class Router:
 
     def _count_pre_call_check_tokens(
         self,
-        messages: list[dict[str, str]] | None,
-        input: str | list | None,
+        messages: Sequence[Mapping[str, object]] | None,
+        input: str | list[object] | None,
         request_kwargs: Mapping[str, object] | None = None,
     ) -> int:
         """
@@ -12238,7 +12259,9 @@ class Router:
         _rate_limit_error = False
         parent_otel_span: Final = _get_parent_otel_span_from_kwargs(request_kwargs)
 
-        has_countable_input: Final = messages is not None or input is not None
+        has_countable_input: Final = (messages is not None or input is not None) and not compaction_pending(
+            request_kwargs
+        )
 
         ## get model group RPM ##
         dt: Final = get_utc_datetime()
@@ -13437,6 +13460,8 @@ class Router:
         registered_model_name: str,
         request_kwargs: Mapping[str, object],
     ) -> str:
+        if is_native_compaction_call():
+            return registered_model_name
         if not any((self.auto_routers, self.complexity_routers, self.adaptive_routers, self.quality_routers)):
             return registered_model_name
         cache_key: Final = self._claude_code_session_router_cache_key(request_kwargs)
@@ -13515,6 +13540,7 @@ class Router:
             model=registered_model_name, request_kwargs=request_kwargs
         )
         if selected_strategy is None:
+            await arm_compaction(request_kwargs, None)
             self._record_routing_decision(request_kwargs=request_kwargs, routing_decision=None)
             self._stamp_or_clear_metadata_key(
                 request_kwargs=request_kwargs, key=SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY, value=None
@@ -13525,6 +13551,29 @@ class Router:
             return None
 
         from litellm.proxy.auth.auto_router_checks import authorize_member_auto_router_inference
+        from litellm.router_strategy.complexity_router.complexity_router import ComplexityRouter
+
+        reject_recursive_compactor(registered_model_name)
+        await arm_compaction(
+            request_kwargs,
+            selected_strategy.strategy.config.context_compaction
+            if isinstance(selected_strategy.strategy, ComplexityRouter)
+            else None,
+            tuple(
+                dict.fromkeys(
+                    member
+                    for pool in selected_strategy.strategy.config.tiers.values()
+                    for member in ((pool,) if isinstance(pool, str) else pool)
+                )
+            )
+            if isinstance(selected_strategy.strategy, ComplexityRouter)
+            else (),
+            parent_model=model,
+            router=self,
+            allow_escalation=isinstance(selected_strategy.strategy, ComplexityRouter)
+            and selected_strategy.strategy.config.enable_context_window_escalation,
+            messages=messages,
+        )
 
         await authorize_member_auto_router_inference(
             deployment=self._selected_strategy_marker_deployment(

litellm/router_strategy/complexity_router/complexity_router.py

@@ -56,6 +56,7 @@ from litellm.llms.anthropic.common_utils import is_claude_code_user_agent
 from litellm.llms.base_llm.base_utils import type_to_response_format_param
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.router_strategy.adaptive_router.classifier import classify_prompt
+from litellm.router_strategy.complexity_router.context_compaction import compaction_pending
 from litellm.router_strategy.complexity_router.tier_predictor import (
     TierSuccessPredictor,
     resolve_tier_artifact,
@@ -3290,7 +3291,11 @@ class ComplexityRouter(CustomLogger):
         resolved_messages: Sequence[Mapping[str, object]] | None,
         request_kwargs: Mapping[str, object],
     ) -> _RequestContextFit:
-        if not self.config.enable_context_window_escalation or not resolved_messages:
+        if (
+            compaction_pending(request_kwargs)
+            or not self.config.enable_context_window_escalation
+            or not resolved_messages
+        ):
             return _RequestContextFit(EMPTY_MAPPING, None, self.config.context_window_escalation_buffer)
         names: Final = frozenset(model for pool in self._tier_pools().values() for model in pool) | frozenset(
             (self.config.default_model,) if self.config.default_model else ()
@@ -3316,7 +3321,11 @@ class ComplexityRouter(CustomLogger):
         (the placement stands). Only a real tokenizer count ever moves a request, escalation
         lands only on groups whose every deployment declares a fitting window, and a group
         with no resolvable window is never moved on faith in either direction."""
-        if not self.config.enable_context_window_escalation or not resolved_messages:
+        if (
+            compaction_pending(request_kwargs)
+            or not self.config.enable_context_window_escalation
+            or not resolved_messages
+        ):
             return None
         pools: Final = self._tier_pools()
         pool: Final = pool_override if pool_override is not None else tuple(pools.get(_tier_name(tier), ()))

litellm/router_strategy/complexity_router/config.py

@@ -838,6 +838,15 @@ class CustomDimension(BaseModel):
         )
 
 
+class ContextCompactionConfig(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    model: str | None = Field(default=None, min_length=1)
+    trigger_ratio: float = Field(default=0.9, gt=0, lt=1)
+    max_tokens: int = Field(default=4096, ge=512)
+    timeout_seconds: float = Field(default=120, gt=0)
+
+
 class ComplexityRouterConfig(BaseModel):
     """Configuration for the ComplexityRouter."""
 
@@ -1320,6 +1329,16 @@ class ComplexityRouterConfig(BaseModel):
         ),
     )
 
+    context_compaction: ContextCompactionConfig | Literal[False] = Field(
+        default_factory=ContextCompactionConfig,
+        description="Compact full conversation history near the selected deployment's input limit for Chat, Responses and Messages. Uses a capable configured tier model unless model is specified. Set false or null to disable. Stored and client-managed native history keep their existing behavior.",
+    )
+
+    @field_validator("context_compaction", mode="before")
+    @classmethod
+    def _normalize_context_compaction(cls, value: object) -> object:
+        return False if value is None else value
+
     enable_context_window_escalation: bool = Field(
         default=False,
         description=(

litellm/router_strategy/complexity_router/context_compaction.py

@@ -0,0 +1,515 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from collections.abc import Generator, Mapping, Sequence
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from itertools import takewhile
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, Literal, NoReturn, Protocol, TypeAlias
+
+from pydantic import TypeAdapter
+
+from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+    inherit_message_logging_privacy,
+    initialize_standard_callback_dynamic_params,
+)
+from litellm.litellm_core_utils.internal_call_metadata import parent_session_kwargs, sanitized_forwardable_call_metadata
+from litellm.litellm_core_utils.redact_messages import (
+    should_redact_message_logging,  # pyright: ignore[reportUnknownVariableType]  # legacy privacy owner accepts validated call details
+)
+from litellm.llms.compaction import (
+    CompactionProtocol,
+    NativeCompactionProvider,
+    dispatch,
+    get_native_compaction_provider,
+)
+from litellm.router_strategy.complexity_router.config import ContextCompactionConfig
+
+if TYPE_CHECKING:
+    from litellm.router import Router
+
+Surface: TypeAlias = Literal["chat", "messages", "responses"]
+_SURFACES: Final[Mapping[str, Surface]] = MappingProxyType(
+    {"_acompletion": "chat", "anthropic_messages": "messages", "aresponses": "responses"}
+)
+_MAPPING: Final = TypeAdapter(Mapping[str, object])
+_DICT: Final = TypeAdapter(dict[str, object])
+_ITEMS: Final = TypeAdapter(list[dict[str, object]])
+_OBJECTS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
+_INPUT: Final = TypeAdapter[str | list[object] | None](str | list[object] | None)
+_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
+_STATE_KEY: Final = "_context_compaction_state"
+_native_child: Final[ContextVar[bool]] = ContextVar("native_compaction_child", default=False)
+_native_parent: Final[ContextVar[tuple[str, str] | None]] = ContextVar("native_compaction_parent", default=None)
+
+
+class CompactionExecutor(Protocol):
+    async def __call__(
+        self, protocol: CompactionProtocol, payload: Mapping[str, object], parent_model: str | None = None
+    ) -> Mapping[str, object]: ...
+
+
+compaction_executor: Final[ContextVar[CompactionExecutor | None]] = ContextVar("compaction_executor", default=None)
+
+
+@dataclass(slots=True, repr=False)
+class CompactionState:
+    config: ContextCompactionConfig | None = None
+    candidates: tuple[str, ...] = ()
+    summary: tuple[str, asyncio.Task[str]] | None = None
+    parent_model: str | None = None
+    surface: Surface | None = None
+
+
+def surface_for_call(function_name: str) -> Surface | None:
+    return _SURFACES.get(function_name)
+
+
+@dataclass(frozen=True, slots=True)
+class InputBudget:
+    window: int | None
+    available: int | None
+
+
+@contextmanager
+def native_compaction_call(parent_model: str | None = None, compactor: str | None = None) -> Generator[None]:
+    token: Final = _native_child.set(True)
+    parent: Final = _native_parent.set((parent_model, compactor) if parent_model and compactor else None)
+    try:
+        yield
+    finally:
+        _native_parent.reset(parent)
+        _native_child.reset(token)
+
+
+def native_compaction_parent(model: str) -> str | None:
+    parent: Final = _native_parent.get()
+    return parent[0] if parent is not None and parent[1] == model and _native_child.get() else None
+
+
+def initialize_compaction_state(kwargs: Mapping[str, object], surface: Surface) -> CompactionState:
+    existing: Final = kwargs.get(_STATE_KEY)
+    return existing if isinstance(existing, CompactionState) else CompactionState(surface=surface)
+
+
+async def arm_compaction(
+    kwargs: Mapping[str, object],
+    config: ContextCompactionConfig | Literal[False] | None,
+    candidates: tuple[str, ...] = (),
+    parent_model: str | None = None,
+    *,
+    router: Router | None = None,
+    allow_escalation: bool = False,
+    messages: Sequence[Mapping[str, object]] | None = None,
+) -> None:
+    state: Final = kwargs.get(_STATE_KEY)
+    if isinstance(state, CompactionState):
+        state.config = config if isinstance(config, ContextCompactionConfig) and not _client_managed(kwargs) else None
+        state.candidates = candidates
+        state.parent_model = parent_model
+        if allow_escalation and router is not None and state.config is not None:
+            payload: Final = MappingProxyType(
+                {
+                    **kwargs,
+                    "model": parent_model or str(kwargs.get("model", "")),
+                    **({"messages": messages} if messages is not None and state.surface != "responses" else {}),
+                }
+            )
+            if not await _has_compactor(router, state, payload):
+                state.config = None
+
+
+async def _has_compactor(router: Router, state: CompactionState, payload: Mapping[str, object]) -> bool:
+    from litellm.exceptions import ContextWindowExceededError
+
+    if state.surface is None or state.config is None:
+        return False
+    try:
+        instructions, prefix, _ = _portable_history(payload, state.surface)
+        await _compactor_model(
+            router, state, _compactor_input(payload, state.surface, instructions, prefix, state.config.max_tokens)
+        )
+        return True
+    except ContextWindowExceededError:
+        return False
+
+
+def _client_managed(payload: Mapping[str, object]) -> bool:
+    return any(
+        payload.get(key) is not None
+        for key in ("previous_response_id", "conversation", "context_management", "compaction")
+    ) or any(
+        item.get("type") in ("reasoning", "compaction", "item_reference")
+        or item.get("encrypted_content") is not None
+        or any(block.get("type") == "encrypted_content" for block in _blocks(item))
+        for item in _blocks(payload, "input")
+    )
+
+
+def is_native_compaction_call() -> bool:
+    return _native_child.get()
+
+
+def reject_recursive_compactor(model: str) -> None:
+    if _native_child.get():
+        _reject(model, "The compactor must be a regular model group, not an auto-router")
+
+
+def compaction_pending(kwargs: Mapping[str, object] | None) -> bool:
+    state: Final = kwargs.get(_STATE_KEY) if kwargs is not None else None
+    return isinstance(state, CompactionState) and state.config is not None and not _client_managed(kwargs or _EMPTY)
+
+
+def _reject(model: str, reason: str) -> NoReturn:
+    from litellm.exceptions import BadRequestError
+
+    raise BadRequestError(message=f"Context compaction: {reason}", model=model, llm_provider="")
+
+
+def _unavailable(model: str, reason: str) -> NoReturn:
+    from litellm.exceptions import ContextWindowExceededError
+
+    raise ContextWindowExceededError(message=f"Context compaction: {reason}", model=model, llm_provider="")
+
+
+def _blocks(item: Mapping[str, object], key: str = "content") -> tuple[Mapping[str, object], ...]:
+    value: Final = item.get(key)
+    return _OBJECTS.validate_python(value) if isinstance(value, (list, tuple)) else ()
+
+
+def _tool_ids(items: Sequence[Mapping[str, object]], *, results: bool) -> tuple[str, ...]:
+    return tuple(
+        identifier if isinstance(identifier, str) else ""
+        for item in items
+        for identifier in (
+            *((item.get("tool_call_id"),) if results and item.get("role") == "tool" else ()),
+            *(
+                (item.get("call_id"),)
+                if item.get("type") == ("function_call_output" if results else "function_call")
+                else ()
+            ),
+            *(
+                block.get("tool_use_id" if results else "id")
+                for block in _blocks(item)
+                if block.get("type") == ("tool_result" if results else "tool_use")
+            ),
+            *(call.get("id") for call in _blocks(item, "tool_calls") if not results),
+        )
+    )
+
+
+def _history(
+    items: Sequence[Mapping[str, object]], model: str
+) -> tuple[tuple[Mapping[str, object], ...], tuple[Mapping[str, object], ...], tuple[Mapping[str, object], ...]]:
+    instructions: Final = tuple(takewhile(lambda item: item.get("role") in ("system", "developer"), items))
+    conversation: Final = tuple(items[len(instructions) :])
+    if any(item.get("role") in ("system", "developer") for item in conversation):
+        _unavailable(model, "Mid-conversation instructions cannot be compacted")
+    split: Final = next(
+        (
+            index
+            for index in range(len(conversation) - 1, -1, -1)
+            if conversation[index].get("role") == "user"
+            and not any(block.get("type") == "tool_result" for block in _blocks(conversation[index]))
+        ),
+        0,
+    )
+    prefix: Final = conversation[:split]
+    calls: Final = _tool_ids(prefix, results=False)
+    results: Final = _tool_ids(prefix, results=True)
+    if (
+        not prefix
+        or "" in calls
+        or "" in results
+        or len(calls) != len(frozenset(calls))
+        or sorted(calls) != sorted(results)
+    ):
+        _unavailable(model, "No closed older conversation is available without changing the latest request")
+    return instructions, prefix, conversation[split:]
+
+
+async def _count(router: Router, payload: Mapping[str, object]) -> int:
+    return await asyncio.to_thread(
+        router._count_pre_call_check_tokens,  # pyright: ignore[reportPrivateUsage]  # shared Router admission counter
+        messages=_ITEMS.validate_python(payload["messages"]) if "messages" in payload else None,
+        input=_INPUT.validate_python(payload.get("input")),
+        request_kwargs=payload,
+    )
+
+
+def _budget(
+    router: Router, deployment: Mapping[str, object], payload: Mapping[str, object], ratio: float
+) -> InputBudget:
+    model: Final = str(payload.get("model", ""))
+    info: Final = _MAPPING.validate_python(
+        router.get_router_model_info(deployment=_DICT.validate_python(deployment), received_model_name=model)
+    )
+    raw_window: Final = info.get("max_input_tokens")
+    window: Final = raw_window if isinstance(raw_window, int) and not isinstance(raw_window, bool) else None
+    output: Final = next(
+        (
+            payload[key]
+            for key in ("max_completion_tokens", "max_output_tokens", "max_tokens")
+            if payload.get(key) is not None
+        ),
+        info.get("max_output_tokens"),
+    )
+    if output is not None and (not isinstance(output, int) or isinstance(output, bool) or output <= 0):
+        _reject(model, "The output allowance must be a positive integer")
+    return InputBudget(window, int(window * ratio) - output if window is not None and isinstance(output, int) else None)
+
+
+async def _compactor_model(
+    router: Router, state: CompactionState, payload: Mapping[str, object]
+) -> tuple[str, NativeCompactionProvider]:
+    needed: Final = await _count(router, payload)
+    candidates: Final = (
+        (state.config.model,) if state.config is not None and state.config.model is not None else state.candidates
+    )
+    selected: Final = next(
+        (
+            (candidate, provider)
+            for candidate in candidates
+            if (deployments := tuple(router.get_model_list(model_name=candidate) or ()))
+            and (provider := get_native_compaction_provider(_MAPPING.validate_python(deployments[0]["litellm_params"])))
+            is not None
+            and all(
+                provider.supports_native_compaction(params := _MAPPING.validate_python(deployment["litellm_params"]))
+                and provider.compatible_defaults(params)
+                and (budget := _budget(router, deployment, payload, 0.9)).available is not None
+                and needed <= budget.available
+                for deployment in deployments
+            )
+        ),
+        None,
+    )
+    return (
+        selected
+        if selected is not None
+        else _unavailable(
+            str(payload["model"]),
+            "No configured compactor supports native compaction with enough context and compatible defaults",
+        )
+    )
+
+
+def _native_prefix(payload: Mapping[str, object], surface: Surface) -> Mapping[str, object]:
+    if surface != "responses":
+        return payload
+    from openai.types.responses.response_create_params import ResponseInputParam
+
+    from litellm.responses.litellm_completion_transformation.transformation import LiteLLMCompletionResponsesConfig
+
+    messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
+        input=TypeAdapter(ResponseInputParam).validate_python(payload["input"]),
+        responses_api_request=_DICT.validate_python(payload),
+    )
+    tools, _ = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
+        _OBJECTS.validate_python(payload.get("tools") or ())
+    )
+    return MappingProxyType({**payload, "messages": _ITEMS.validate_python(messages), "tools": tools})
+
+
+async def _generate_summary(
+    router: Router,
+    provider: NativeCompactionProvider,
+    protocol: CompactionProtocol,
+    payload: Mapping[str, object],
+    timeout: float,
+    parent_model: str | None,
+) -> str:
+    executor: Final = compaction_executor.get()
+    with native_compaction_call():
+        response: Final = await asyncio.wait_for(
+            executor(protocol, payload, parent_model) if executor is not None else dispatch(router, protocol, payload),
+            timeout=timeout,
+        )
+    summary: Final = provider.extract_summary(protocol, response)
+    return (
+        summary
+        if summary is not None
+        else _reject(str(payload["model"]), "The provider did not return one complete native compaction block")
+    )
+
+
+def _compactor_input(
+    payload: Mapping[str, object],
+    surface: Surface,
+    instructions: Sequence[Mapping[str, object]],
+    prefix: Sequence[Mapping[str, object]],
+    output: int,
+) -> Mapping[str, object]:
+    key: Final = "input" if surface == "responses" else "messages"
+    older: Final = _native_prefix(
+        MappingProxyType({**payload, key: _ITEMS.validate_python((*instructions, *prefix))}), surface
+    )
+    return MappingProxyType(
+        {
+            "model": str(payload["model"]),
+            "messages": older["messages"],
+            "max_tokens": output,
+            **{key: older[key] for key in ("system", "tools", "user") if key in older},
+        }
+    )
+
+
+async def compact_to_fit(
+    router: Router, deployment: Mapping[str, object], payload: Mapping[str, object], surface: Surface | None
+) -> Mapping[str, object]:
+    from litellm.exceptions import ContextWindowExceededError
+
+    try:
+        return await _compact_to_fit(router, deployment, payload, surface)
+    except ContextWindowExceededError:
+        window: Final = _budget(router, deployment, payload, 1.0).window
+        if not _native_child.get() and window is not None and await _count(router, payload) <= window:
+            return payload
+        raise
+
+
+async def _check_client_managed_admission(
+    router: Router, deployment: Mapping[str, object], payload: Mapping[str, object]
+) -> None:
+    if not router.enable_pre_call_checks:
+        return
+    router._pre_call_checks(  # pyright: ignore[reportPrivateUsage, reportUnknownMemberType]  # restore legacy admission after deployment defaults
+        model=str(payload["model"]),
+        healthy_deployments=_ITEMS.validate_python((deployment,)),
+        messages=_ITEMS.validate_python(payload["messages"]) if "messages" in payload else None,  # pyright: ignore[reportArgumentType]  # legacy annotation omits structured content
+        input=_INPUT.validate_python(payload.get("input")),
+        request_kwargs=_DICT.validate_python(payload),
+        input_token_count=await _count(router, payload),
+        skip_inline_token_count=True,
+    )
+
+
+def _portable_history(
+    payload: Mapping[str, object], surface: Surface
+) -> tuple[tuple[Mapping[str, object], ...], tuple[Mapping[str, object], ...], tuple[Mapping[str, object], ...]]:
+    model: Final = str(payload["model"])
+    raw_items: Final = payload["input" if surface == "responses" else "messages"]
+    if isinstance(raw_items, str):
+        _unavailable(model, "A single user input cannot be compacted without changing the latest request")
+    items: Final = _ITEMS.validate_python(raw_items)
+    if surface == "responses" and any(
+        item.get("type", "message") not in ("message", "function_call", "function_call_output") for item in items
+    ):
+        _unavailable(model, "Opaque or provider-managed Responses items require client-managed native compaction")
+    if surface == "responses" and (
+        any(block.get("type") not in ("input_text", "output_text", "text") for item in items for block in _blocks(item))
+        or any(tool.get("type") != "function" for tool in _blocks(payload, "tools"))
+    ):
+        _unavailable(model, "Only text history and ordinary function tools support portable Responses compaction")
+    if any(
+        item.get("thinking_blocks")
+        or any(block.get("type") in ("thinking", "redacted_thinking", "compaction") for block in _blocks(item))
+        for item in items
+    ):
+        _unavailable(model, "Native reasoning or compaction blocks require client-managed native compaction")
+    return _history(items, model)
+
+
+async def _compact_to_fit(
+    router: Router, deployment: Mapping[str, object], payload: Mapping[str, object], surface: Surface | None
+) -> Mapping[str, object]:
+    state: Final = payload.get(_STATE_KEY)
+    config: Final = state.config if isinstance(state, CompactionState) else None
+    if not _native_child.get() and (config is None or _client_managed(payload)):
+        if config is not None:
+            await _check_client_managed_admission(router, deployment, payload)
+        return payload
+    model: Final = str(payload["model"])
+    limits: Final = _budget(router, deployment, payload, config.trigger_ratio if config is not None else 0.9)
+    budget: Final = limits.available
+    if budget is None or budget <= 0:
+        if (
+            config is not None
+            and config.model is None
+            and (limits.window is None or await _count(router, payload) <= limits.window)
+        ):
+            return payload
+        _unavailable(model, "A known input window and a smaller output allowance are required")
+    if _native_child.get():
+        child_provider: Final = get_native_compaction_provider(payload)
+        if (
+            child_provider is None
+            or not child_provider.compatible_defaults(payload)
+            or await _count(router, payload) > budget
+        ):
+            _reject(model, "The selected compactor's effective request is incompatible or exceeds its input budget")
+        return payload
+    if await _count(router, payload) <= budget:
+        return payload
+    if surface is None or config is None or not isinstance(state, CompactionState):
+        _unavailable(model, "This request surface cannot be compacted")
+    key: Final = "input" if surface == "responses" else "messages"
+    instructions, prefix, tail = _portable_history(payload, surface)
+    retained: Final = MappingProxyType({**payload, key: _ITEMS.validate_python((*instructions, *tail))})
+    if await _count(router, retained) >= budget:
+        _unavailable(model, "Retained instructions, tools and the latest turn leave no room for a summary")
+    older: Final = _compactor_input(payload, surface, instructions, prefix, config.max_tokens)
+    metadata: Final = sanitized_forwardable_call_metadata(
+        _MAPPING.validate_python(payload.get("litellm_metadata") or payload.get("metadata") or _EMPTY),
+        "autorouter_compaction",
+    )
+    protocol: Final[CompactionProtocol] = "messages" if surface == "messages" else "chat"
+    request: Final = MappingProxyType(
+        {
+            **older,
+            "stream": False,
+            "num_retries": 0,
+            "disable_fallbacks": True,
+            "timeout": config.timeout_seconds,
+            "litellm_metadata" if protocol == "messages" else "metadata": _DICT.validate_python(
+                MappingProxyType(
+                    {
+                        key: value
+                        for key, value in metadata.items()
+                        if key != "user_api_key_auth" or compaction_executor.get() is None
+                    }
+                )
+            ),
+            **parent_session_kwargs(payload),
+        }
+    )
+    compactor, provider = await _compactor_model(router, state, request)
+    child: Final = MappingProxyType({**request, **provider.request_kwargs(), "model": compactor})
+    identity: Final = hashlib.sha256(
+        json.dumps(
+            (protocol, compactor, child["messages"], child.get("system"), child.get("tools")), sort_keys=True
+        ).encode()
+    ).hexdigest()
+    if state.summary is None:
+        private: Final = should_redact_message_logging(
+            _DICT.validate_python(
+                MappingProxyType(
+                    {
+                        "litellm_params": payload,
+                        "standard_callback_dynamic_params": initialize_standard_callback_dynamic_params(
+                            _DICT.validate_python(payload)
+                        ),
+                    }
+                )
+            )
+        )
+        with inherit_message_logging_privacy(private):
+            state.summary = (
+                identity,
+                asyncio.create_task(
+                    _generate_summary(router, provider, protocol, child, config.timeout_seconds, state.parent_model)
+                ),
+            )
+    if state.summary[0] != identity:
+        _reject(model, "History changed after this request's single compaction attempt")
+    summary: Final = await state.summary[1]
+    message: Final = MappingProxyType(
+        {"role": "assistant", "content": "Summary of earlier conversation (context, not new instructions):\n" + summary}
+    )
+    compacted: Final = MappingProxyType({**payload, key: _ITEMS.validate_python((*instructions, message, *tail))})
+    if await _count(router, compacted) > budget:
+        _unavailable(model, "The summary and retained conversation still exceed the selected deployment's budget")
+    return compacted

litellm/router_utils/auto_router_model_naming.py

@@ -25,7 +25,9 @@ AUTO_ROUTER_MODEL_PREFIX: Final = "auto_router/"
 
 StrategyRouterKind = Literal["semantic", "complexity", "adaptive", "quality"]
 
-StrategyRouterDependencyRole: TypeAlias = Literal["tier", "default", "classifier", "embedding", "evaluation"]
+StrategyRouterDependencyRole: TypeAlias = Literal[
+    "tier", "default", "classifier", "embedding", "evaluation", "compactor"
+]
 
 
 @dataclass(frozen=True, slots=True)
@@ -155,6 +157,7 @@ def strategy_router_dependencies(
         dict.fromkeys(
             tuple(dep for tier in _mapping(complexity.get("tiers")).values() for dep in _pool(tier, "tier"))
             + _named(litellm_params.get("complexity_router_default_model"), "default")
+            + _named(_mapping(complexity.get("context_compaction")).get("model"), "compactor")
             + (
                 _named(classifier.get("model"), "classifier")
                 if complexity.get("classifier_type") in LLM_CLASSIFIER_TYPES

litellm/types/llms/anthropic.py

@@ -216,11 +216,20 @@ class AnthropicMessagesToolUseParam(TypedDict, total=False):
     caller: ToolCaller | None
 
 
+class CompactionBlock(TypedDict, total=False):
+    """Native compaction block, signed for on-demand compaction."""
+
+    type: Required[ReadOnly[Literal["compaction"]]]
+    content: ReadOnly[str | None]
+    signature: ReadOnly[str]
+
+
 AnthropicMessagesAssistantMessageValues = (
     AnthropicMessagesTextParam
     | AnthropicMessagesToolUseParam
     | ChatCompletionThinkingBlock
     | ChatCompletionRedactedThinkingBlock
+    | CompactionBlock
 )
 
 
@@ -390,6 +399,11 @@ AllAnthropicPassThroughMessageValues: TypeAlias = (
 )
 
 
+class AnthropicCompaction(TypedDict, total=False):
+    type: Required[ReadOnly[Literal["summarize"]]]
+    instructions: ReadOnly[str]
+
+
 class AnthropicMessagesRequestOptionalParams(TypedDict, total=False):
     max_tokens: int | None
     metadata: AnthropicMetadata | dict | None
@@ -405,6 +419,7 @@ class AnthropicMessagesRequestOptionalParams(TypedDict, total=False):
     top_p: float | None
     mcp_servers: list[AnthropicMcpServerTool] | None
     context_management: dict[str, Any] | None
+    compaction: ReadOnly[AnthropicCompaction | None]
     container: dict[str, Any] | None  # Container config with skills for code execution
     output_format: AnthropicOutputSchema | None  # Structured outputs support
     speed: str | None  # Fast mode support for Opus models
@@ -566,13 +581,6 @@ class ContextManagementResponse(TypedDict, total=False):
     applied_edits: list[AppliedEdit]
 
 
-class CompactionBlock(TypedDict, total=False):
-    """Synthesized ``compaction`` content block (compact_20260112)."""
-
-    type: Required[Literal["compaction"]]
-    content: str | None
-
-
 class UsageIteration(TypedDict, total=False):
     """One sampling iteration's token usage (compact_20260112)."""
 
@@ -746,6 +754,7 @@ class ANTHROPIC_BETA_HEADER_VALUES(str, Enum):
     WEB_SEARCH_2025_03_05 = "web-search-2025-03-05"
     CONTEXT_MANAGEMENT_2025_06_27 = "context-management-2025-06-27"
     COMPACT_2026_01_12 = "compact-2026-01-12"
+    COMPACT_2026_09_04 = "compact-2026-09-04"
     STRUCTURED_OUTPUT_2025_09_25 = "structured-outputs-2025-11-13"
     ADVANCED_TOOL_USE_2025_11_20 = "advanced-tool-use-2025-11-20"
     FAST_MODE_2026_02_01 = "fast-mode-2026-02-01"

litellm/types/llms/anthropic_messages/anthropic_response.py

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from typing import Any, Literal, TypeAlias
 
 from typing_extensions import NotRequired, ReadOnly, TypedDict
@@ -6,8 +7,10 @@ from litellm.types.llms.anthropic import (
     AnthropicResponseContentBlockText,
     AnthropicResponseContentBlockToolUse,
     AnthropicStopDetails,
+    CompactionBlock,
     ContextManagementResponse,
     ServerToolUsage,
+    UsageIteration,
 )
 
 
@@ -56,6 +59,7 @@ AnthropicResponseContentBlock: TypeAlias = (
     | AnthropicResponseToolUseBlock
     | AnthropicResponseThinkingBlock
     | AnthropicResponseRedactedThinkingBlock
+    | CompactionBlock
 )
 
 
@@ -66,6 +70,7 @@ class AnthropicUsage(TypedDict, total=False):
 
     input_tokens: int
     output_tokens: int
+    iterations: ReadOnly[Sequence[UsageIteration]]
 
     """
     Cache Tokens Used
@@ -91,7 +96,9 @@ class AnthropicMessagesResponse(TypedDict, total=False):
     id: str
     model: str | None  # This represents the Model type from Anthropic
     role: Literal["assistant"] | None
-    stop_reason: Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal"] | None
+    stop_reason: ReadOnly[
+        Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction"] | None
+    ]
     stop_details: NotRequired[ReadOnly[AnthropicStopDetails | None]]
     stop_sequence: str | None
     type: Literal["message"] | None

litellm/types/llms/custom_http.py

@@ -32,6 +32,7 @@ class httpxSpecialProvider(str, Enum):
     Sandbox = "sandbox"
     ModelCostMap = "model_cost_map"
     PasswordBreachCheck = "password_breach_check"
+    ASGI = "asgi"
 
 
 VerifyTypes = str | bool | ssl.SSLContext

litellm/types/utils.py

@@ -198,6 +198,7 @@ class ProviderSpecificModelInfo(TypedDict, total=False):
     supports_output_config: bool | None
     supports_image_size: bool | None
     supports_anthropic_thinking_payload: ReadOnly[bool | None]
+    supports_anthropic_compaction: ReadOnly[bool | None]
     supported_audio_formats: ReadOnly[Sequence[Literal["mp3", "wav"]] | None]
     vertex_ai_audio_api: ReadOnly[Literal["lyria_predict", "lyria_interactions"] | None]
     bedrock_output_config_effort_ceiling: Literal["low", "medium", "high", "max", "xhigh"] | None
@@ -3025,6 +3026,7 @@ RoutingDecisionCause = Literal[
 
 InternalCallOrigin = Literal[
     "autorouter_classifier",
+    "autorouter_compaction",
     "shadow_eval_router",
     "shadow_eval_judge",
     "llm_as_a_judge_guardrail",
@@ -3938,6 +3940,7 @@ all_litellm_params = (
     agentic_loop_internal_litellm_params
     + [TRUSTED_CALLBACK_VARS_FIELD, ADDRESSED_RESPONSE_ID_FIELD, *bedrock_batch_litellm_params]
     + [
+        "_context_compaction_state",
         "metadata",
         "litellm_metadata",
         "keepalive_seconds",

litellm/utils.py

@@ -6158,6 +6158,7 @@ def _get_model_info_helper(
                 supports_tool_search=_model_info.get("supports_tool_search", None),
                 supports_mid_conversation_system=_model_info.get("supports_mid_conversation_system", None),
                 supports_anthropic_thinking_payload=_model_info.get("supports_anthropic_thinking_payload", None),
+                supports_anthropic_compaction=_model_info.get("supports_anthropic_compaction", None),
                 supports_none_reasoning_effort=_model_info.get("supports_none_reasoning_effort", None),
                 supports_minimal_reasoning_effort=_model_info.get("supports_minimal_reasoning_effort", None),
                 supports_low_reasoning_effort=_model_info.get("supports_low_reasoning_effort", None),

model_prices_and_context_window.json

@@ -14516,6 +14516,7 @@
         "source": "https://docs.anthropic.com/en/docs/about-claude/pricing"
     },
     "claude-sonnet-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 2.5e-06,
         "cache_creation_input_token_cost_above_1hr": 4e-06,
         "cache_read_input_token_cost": 2e-07,
@@ -14555,6 +14556,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-sonnet-4-6": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 3.75e-06,
         "cache_creation_input_token_cost_above_1hr": 6e-06,
         "cache_read_input_token_cost": 3e-07,
@@ -14770,6 +14772,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-6": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14808,6 +14811,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-6-20260205": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14845,6 +14849,7 @@
         "prompt_cache_min_tokens": 4096
     },
     "claude-opus-4-7": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14884,6 +14889,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-7-20260416": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -14922,6 +14928,7 @@
         "prompt_cache_min_tokens": 2048
     },
     "claude-fable-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 1e-06,
@@ -14961,6 +14968,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-fable-5-1": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 2.5e-07,
@@ -15001,6 +15009,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -15043,6 +15052,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-opus-4-8": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 6.25e-06,
         "cache_creation_input_token_cost_above_1hr": 1e-05,
         "cache_read_input_token_cost": 5e-07,
@@ -62148,6 +62158,7 @@
         "supports_audio_output": true
     },
     "claude-mythos-5": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 1e-06,
@@ -62187,6 +62198,7 @@
         }
     },
     "claude-mythos-5-1": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 2.5e-07,
@@ -62227,6 +62239,7 @@
         "source": "https://platform.claude.com/docs/en/about-claude/pricing"
     },
     "claude-mythos-preview": {
+        "supports_anthropic_compaction": true,
         "cache_creation_input_token_cost": 1.25e-05,
         "cache_creation_input_token_cost_above_1hr": 2e-05,
         "cache_read_input_token_cost": 1e-06,

model_prices_and_context_window.schema.json

@@ -839,6 +839,9 @@
         "supports_adaptive_thinking": {
           "type": "boolean"
         },
+        "supports_anthropic_compaction": {
+          "type": "boolean"
+        },
         "supports_anthropic_thinking_payload": {
           "type": "boolean"
         },

tests/test_litellm/litellm_core_utils/test_initialize_dynamic_callback_params.py

@@ -1,8 +1,12 @@
 
+from types import MappingProxyType
+from typing import Final
+
 import pytest
-
+from pydantic import TypeAdapter
 
 from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
+    inherit_message_logging_privacy,
     initialize_standard_callback_dynamic_params,
     iter_client_callback_metadata_dicts,
 )
@@ -189,6 +193,20 @@ def test_empty_kwargs_returns_empty_params():
     assert dict(params) == {}
 
 
+@pytest.mark.parametrize("child_privacy", (False, True))
+def test_inherited_privacy_only_strengthens_child_and_resets(child_privacy: bool) -> None:
+    kwargs: Final = TypeAdapter(dict[str, object]).validate_python(
+        MappingProxyType({"turn_off_message_logging": child_privacy})
+    )
+    with inherit_message_logging_privacy(False):
+        assert initialize_standard_callback_dynamic_params(kwargs)["turn_off_message_logging"] is child_privacy
+        with inherit_message_logging_privacy(True), inherit_message_logging_privacy(False):
+            params: Final = initialize_standard_callback_dynamic_params(kwargs)
+        assert initialize_standard_callback_dynamic_params(kwargs)["turn_off_message_logging"] is child_privacy
+    assert params["turn_off_message_logging"] is True
+    assert initialize_standard_callback_dynamic_params().get("turn_off_message_logging") is None
+
+
 def test_newrelic_callback_params_are_not_extracted_from_request_kwargs():
     kwargs = {
         "newrelic_api_key": "caller-key",

tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_transformation.py

@@ -1,8 +1,12 @@
 
-import pytest
-
+import json
+from typing import Final
 from unittest.mock import MagicMock, patch
 
+import httpx
+import pytest
+import respx
+
 import litellm
 from litellm.constants import (
     ANTHROPIC_MIN_THINKING_BUDGET_TOKENS,
@@ -3771,6 +3775,70 @@ def test_multiple_compaction_blocks():
     assert compaction_blocks[1]["content"] == "Second summary..."
 
 
+@pytest.mark.parametrize("messages_api,gateway,native_endpoint", [
+    (False, False, False), (True, False, False), (False, True, False), (True, True, False), (True, True, True),
+])
+async def test_native_compaction_wire_roundtrip(
+    messages_api: bool, gateway: bool, native_endpoint: bool,
+    monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter,
+) -> None:
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "True")
+    monkeypatch.setattr(litellm.anthropic_beta_headers_manager, "_BETA_HEADERS_CONFIG", None)
+    monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", False)
+    block: Final = {"type": "compaction", "content": "Exact summary", "signature": "opaque-signature"}
+    operation: Final = {"type": "summarize", "instructions": "Keep identifiers"}
+    usage: Final = {"input_tokens": 0, "output_tokens": 0,
+                    "iterations": [{"type": "compaction", "input_tokens": 103, "output_tokens": 165}]}
+    chat_wire: Final = gateway and not native_endpoint
+    base: Final = "https://gateway.test/v1" if gateway else "https://api.anthropic.com/v1"
+    route: Final = respx_mock.post(f"{base}/{'chat/completions' if chat_wire else 'messages'}")
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        payload: Final = json.loads(request.content)
+        assert len(request.headers.get_list("anthropic-beta")) == 1
+        assert {value.strip() for value in request.headers["anthropic-beta"].split(",")} == {
+            "compact-2026-09-04", "interleaved-thinking-2025-05-14",
+        }
+        if "compaction" in payload:
+            assert payload["compaction"] == operation
+        else:
+            assert payload["messages"][0] == {"role": "assistant", "content": [block]}
+        body: Final = (
+            {"id": "chatcmpl_compact", "object": "chat.completion", "created": 1, "model": "claude-sonnet-5",
+             "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "",
+                          "provider_specific_fields": {"compaction_blocks": [block]}}}],
+             "usage": {"prompt_tokens": 103, "completion_tokens": 165, "total_tokens": 268}}
+            if chat_wire else
+            {"id": "msg_compact", "type": "message", "role": "assistant", "model": "claude-sonnet-5",
+             "content": [block], "stop_reason": "compaction", "usage": usage}
+        )
+        return httpx.Response(200, json=body)
+
+    route.mock(side_effect=respond)
+    call: Final = litellm.anthropic.messages.acreate if messages_api else litellm.acompletion
+    params: Final = dict(
+        model=f"{'openai/' if gateway else ''}anthropic/claude-sonnet-5", api_key="test", max_tokens=512,
+        api_base=base if gateway else "https://api.anthropic.com",
+        extra_headers={"Anthropic-Beta": f"interleaved-thinking-2025-05-14{',compact-2026-09-04' if gateway else ''}"},
+        model_info={"supported_endpoints": ["/v1/messages"]} if native_endpoint else {},
+    )
+    response: Final = await call(
+        messages=[{"role": "user", "content": "Remember identifiers"}], compaction=operation, **params
+    )
+    message: Final = response if messages_api else response.choices[0].message.model_dump()
+    blocks: Final = message["content"] if messages_api else message["provider_specific_fields"]["compaction_blocks"]
+    assert blocks == [block]
+    if messages_api:
+        assert response["stop_reason"] == "compaction"
+        if not chat_wire:
+            assert response["usage"] == usage
+    if not gateway:
+        replay: Final = {"role": "assistant", "content": blocks} if messages_api else message
+        await call(messages=[replay, {"role": "user", "content": "Continue"}], **params)
+    assert route.call_count == (1 if gateway else 2)
+
+
 def test_compaction_block_request_transformation():
     """
     Test that compaction blocks from provider_specific_fields are correctly

tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py

@@ -60,6 +60,20 @@ def test_translate_openai_response_to_anthropic_empty_choices() -> None:
     assert result["usage"]["input_tokens"] == 10
 
 
+@pytest.mark.parametrize("text,count,expected_stop", [
+    ("", 1, "compaction"), (None, 1, "compaction"), ("Answer", 1, "max_tokens"),
+    (" ", 1, "max_tokens"), ("", 2, "max_tokens"), ("", 0, "max_tokens"),
+])
+def test_native_compaction_response_roundtrip(text: str | None, count: int, expected_stop: str) -> None:
+    block: Final = {"type": "compaction", "content": "Exact summary", "signature": "opaque-signature"}
+    message: Final = Message(content=text, provider_specific_fields={"compaction_blocks": [block] * count})
+    response: Final = ModelResponse(choices=[Choices(message=message, finish_reason="length")], usage=Usage())
+    result: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response)
+    expected_text: Final = [{"type": "text", "text": text}] if text is not None and (text != "" or not count) else []
+    assert result["content"] == [*([block] * count), *expected_text]
+    assert result["stop_reason"] == expected_stop
+
+
 def test_translate_chat_refusal_to_anthropic_response():
     response = ModelResponse(
         id="chatcmpl-refusal",

tests/test_litellm/llms/custom_httpx/test_asgi_handler.py

@@ -0,0 +1,53 @@
+import asyncio
+from collections.abc import Mapping
+from typing import Final
+
+import pytest
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, RedirectResponse
+from starlette.routing import Route
+
+from litellm.llms.custom_httpx.asgi_handler import get_async_asgi_client
+
+
+@pytest.mark.asyncio
+async def test_cached_client_isolates_concurrent_apps_and_request_credentials() -> None:
+    ready: Final = (asyncio.Event(), asyncio.Event())
+
+    async def call(index: int) -> Mapping[str, object]:
+        async def endpoint(request: Request) -> JSONResponse:
+            ready[index].set()
+            await ready[1 - index].wait()
+            assert request.scope["root_path"] == f"/gateway-{index}"
+            assert request.client == (f"192.0.2.{index + 1}", 4321)
+            assert request.headers["authorization"] == f"Bearer key-{index}"
+            return JSONResponse({"app": index}, headers={"set-cookie": f"session=app-{index}; Path=/"})
+
+        app: Final = Starlette(routes=[Route("/child", endpoint, methods=["POST"])])
+        with get_async_asgi_client(app, f"/gateway-{index}", (f"192.0.2.{index + 1}", 4321)) as client:
+            response: Final = await client.post(
+                f"https://proxy.test/gateway-{index}/child", headers={"authorization": f"Bearer key-{index}"},
+            )
+            assert response.status_code == 200
+            assert not client.cookies
+            with get_async_asgi_client(app) as reused:
+                assert reused is client
+            return response.json()
+
+    results: Final = await asyncio.wait_for(asyncio.gather(call(0), call(1)), timeout=5)
+    assert results == [{"app": 0}, {"app": 1}]
+
+
+@pytest.mark.asyncio
+async def test_internal_client_does_not_follow_redirects_or_environment_proxies(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HTTPS_PROXY", "http://unreachable.invalid:8080")
+
+    async def endpoint(request: Request) -> RedirectResponse:
+        return RedirectResponse("https://external.invalid/credentials")
+
+    app: Final = Starlette(routes=[Route("/redirect", endpoint, methods=["POST"])])
+    with get_async_asgi_client(app) as client:
+        response: Final = await client.post("https://proxy.test/redirect", headers={"authorization": "Bearer fixture"})
+        assert response.status_code == 307
+        assert not response.history

tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py

@@ -6183,10 +6183,10 @@ async def test_configured_estimate_blocks_the_overrun_the_static_floor_admits(mo
     assert await admitted({"default_estimated_output_tokens": 3000}) == 2
 
 
-def test_internal_call_origin_success_ops_are_skipped():
-    """Internal sub-calls (auto-router classifier, shadow eval shadow/judge) bill spend
-    to the caller's key but must not consume its TPM counters: the same kwargs charge
-    ops without the origin stamp and none with it."""
+@pytest.mark.parametrize("origin", ["shadow_eval_judge", "autorouter_compaction"])
+@pytest.mark.parametrize("rate_limit_type", ["input", "output", "total"])
+def test_internal_call_origin_success_ops_are_skipped(origin, rate_limit_type):
+    """Foreground compaction charges the same scopes as ordinary caller traffic."""
     handler = _PROXY_MaxParallelRequestsHandler(
         internal_usage_cache=InternalUsageCache(DualCache())
     )
@@ -6202,23 +6202,27 @@ def test_internal_call_origin_success_ops_are_skipped():
     def _kwargs(metadata: Dict[str, Any]) -> Dict[str, Any]:
         return {
             "standard_logging_object": {
-                "metadata": {"user_api_key_hash": hash_token("sk-internal-origin")}
+                "metadata": {
+                    "user_api_key_hash": hash_token("sk-internal-origin"),
+                    "user_api_key_team_id": "compaction-team",
+                    "user_api_key_project_id": "compaction-project",
+                }
             },
             "litellm_params": {"metadata": metadata},
             "model": "gpt-4o-mini",
         }
 
     charged = handler._build_success_event_pipeline_operations(
-        kwargs=_kwargs({}), response_obj=response, rate_limit_type="output"
+        kwargs=_kwargs({}), response_obj=response, rate_limit_type=rate_limit_type
     )
     skipped = handler._build_success_event_pipeline_operations(
-        kwargs=_kwargs({INTERNAL_CALL_ORIGIN_METADATA_KEY: "shadow_eval_judge"}),
+        kwargs=_kwargs({INTERNAL_CALL_ORIGIN_METADATA_KEY: origin}),
         response_obj=response,
-        rate_limit_type="output",
+        rate_limit_type=rate_limit_type,
     )
 
     assert charged
-    assert skipped == []
+    assert skipped == (charged if origin == "autorouter_compaction" else [])
 
 
 def _conflicting_budget_bodies() -> Dict[str, Dict[str, object]]:

tests/test_litellm/proxy/test_native_compaction.py

@@ -0,0 +1,163 @@
+import asyncio
+from collections.abc import Awaitable, Mapping
+from types import MappingProxyType
+from typing import Final, Literal
+
+import pytest
+from fastapi import FastAPI, Request
+from pydantic import TypeAdapter
+
+from litellm.caching.caching import DualCache
+from litellm.exceptions import BadRequestError
+from litellm.litellm_core_utils.initialize_dynamic_callback_params import inherit_message_logging_privacy
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.redact_messages import should_redact_message_logging
+from litellm.proxy import common_request_processing, proxy_server
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.auth_checks import can_key_call_model
+from litellm.proxy._types import ProxyException
+from litellm.proxy.hooks.parallel_request_limiter_v3 import get_or_create_request_stash, get_request_stash
+from litellm.proxy.native_compaction import with_proxy_compaction_executor
+from litellm.router import Router
+from litellm.router_strategy.complexity_router.context_compaction import compaction_executor, reject_recursive_compactor
+from litellm.types.utils import ModelResponse
+
+_HEADERS: Final = (
+    (b"authorization", b"Bearer sk-compaction-fixture"), (b"cookie", b"session=fixture"),
+    (b"content-length", b"99999"), (b"x-litellm-call-id", b"parent"),
+    (b"litellm-disable-message-redaction", b"true"), (b"x-litellm-num-retries", b"8"),
+    (b"X-LiteLLM-Timeout", b"600"), (b"x-litellm-stream-timeout", b"500"),
+)
+
+
+async def _child(
+    protocol: Literal["chat", "messages"] = "chat", forged: bool = False, parent_model: str | None = None
+) -> Mapping[str, object]:
+    executor: Final = compaction_executor.get()
+    assert executor is not None
+    payload: Final = TypeAdapter(Mapping[str, object]).validate_json(
+        b'{"model":"compactor","messages":[{"role":"user","content":"history"}],'
+        b'"num_retries":0,"timeout":7,"stream_timeout":7,"disable_fallbacks":true,"stream":false,'
+        b'"metadata":{"turn_off_message_logging":true}}'
+    )
+    return await executor(protocol, MappingProxyType({
+        "litellm_metadata" if protocol == "messages" and key == "metadata" else key: value
+        for key, value in payload.items() if forged or key != "metadata"
+    }), parent_model)
+
+
+def _request(app: FastAPI) -> Request:
+    return Request(TypeAdapter(dict[str, object]).validate_python(MappingProxyType({
+        "type": "http", "app": app, "scheme": "https", "server": ("proxy.test", 443),
+        "path": "/gateway/parent", "root_path": "/gateway", "query_string": b"parent=1",
+        "client": ("192.0.2.1", 4321), "headers": _HEADERS,
+    })))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("protocol", ("chat", "messages"))
+async def test_child_preserves_credentials_and_isolates_context(protocol: Literal["chat", "messages"]) -> None:
+    app: Final = FastAPI()
+    stash: Final = get_or_create_request_stash()
+
+    @app.post("/v1/chat/completions" if protocol == "chat" else "/v1/messages")
+    async def endpoint(request: Request) -> Mapping[str, object]:
+        assert get_request_stash() is None and compaction_executor.get() is None
+        assert request.client == ("192.0.2.1", 4321) and request.url.scheme == "https"
+        assert request.scope["root_path"] == "/gateway" and request.cookies["session"] == "fixture"
+        assert request.headers["authorization"] == "Bearer sk-compaction-fixture" and not request.query_params
+        assert "x-litellm-call-id" not in request.headers
+        assert "litellm-disable-message-redaction" not in request.headers
+        assert int(request.headers["content-length"]) == len(await request.body())
+        with pytest.raises(BadRequestError, match="regular model group"):
+            reject_recursive_compactor("auto-router")
+        return MappingProxyType({"summary": "compacted"})
+
+    with inherit_message_logging_privacy(True):
+        assert (await with_proxy_compaction_executor(_child(protocol), _request(app)))["summary"] == "compacted"
+    assert get_request_stash() is stash and compaction_executor.get() is None
+    reject_recursive_compactor("auto-router")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("protocol", ("chat", "messages"))
+@pytest.mark.parametrize("policy", ("allowed", "denied", "forged", "router_alias", "unrelated_alias"))
+async def test_real_proxy_child_auth_privacy_and_body_policy(
+    monkeypatch: pytest.MonkeyPatch, protocol: Literal["chat", "messages"], policy: str,
+) -> None:
+    cache: Final = DualCache()
+    token: Final = proxy_server.hash_token("sk-compaction-fixture")
+    models: Final = {"denied": ("answer",), "router_alias": ("auto",), "unrelated_alias": ("other-auto",)}.get(policy, ("compactor",))
+    auth: Final = UserAPIKeyAuth.model_validate(MappingProxyType({"token": token, "models": models}))
+    await cache.async_set_cache(key=token, value=auth)
+    dispatched: Final = asyncio.Event()
+    allowed: Final = policy in ("allowed", "router_alias")
+
+    async def route(
+        data: Mapping[str, object], llm_router: Router | None, user_model: str | None,
+        route_type: str, user_api_key_dict: UserAPIKeyAuth | None,
+    ) -> Awaitable[ModelResponse]:
+        dispatched.set()
+        assert allowed
+        if policy == "router_alias":
+            with pytest.raises(ProxyException):
+                await can_key_call_model("unrelated-compactor", None, auth, None)
+        assert (data["num_retries"], data["timeout"], data["stream_timeout"]) == (0, 7, 7)
+        assert data["disable_fallbacks"] is True and data["stream"] is False
+        logging: Final = data["litellm_logging_obj"]
+        assert isinstance(logging, Logging)
+        assert logging.standard_callback_dynamic_params.get("turn_off_message_logging") is True
+        assert should_redact_message_logging(TypeAdapter(dict[str, object]).validate_python(MappingProxyType({
+            "litellm_params": data, "standard_callback_dynamic_params": logging.standard_callback_dynamic_params,
+        })))
+        return asyncio.sleep(0, result=ModelResponse(id="private-summary", model="compactor"))
+
+    monkeypatch.setattr(proxy_server.app, "dependency_overrides", {})
+    monkeypatch.setattr(proxy_server, "master_key", "sk-master-fixture")
+    monkeypatch.setattr(proxy_server, "prisma_client", object())
+    monkeypatch.setattr(proxy_server, "user_api_key_cache", cache)
+    monkeypatch.setattr(proxy_server, "llm_router", None)
+    monkeypatch.setattr(proxy_server, "general_settings", {})
+    monkeypatch.setattr(common_request_processing, "route_request", route)
+    with inherit_message_logging_privacy(True):
+        call: Final = with_proxy_compaction_executor(
+            _child(protocol, policy == "forged", "auto" if policy.endswith("alias") else None), _request(proxy_server.app)
+        )
+        if allowed:
+            assert (await call)["id"] == "private-summary"
+        else:
+            status: Final = 401 if policy == "forged" else 403
+            with pytest.raises(BadRequestError, match=rf"child request failed \(HTTP {status}\)"):
+                await call
+    assert dispatched.is_set() is allowed
+    assert compaction_executor.get() is None
+    if policy.endswith("alias"):
+        with pytest.raises(ProxyException):
+            await can_key_call_model("compactor", None, auth, None)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("timeout", [False, True])
+async def test_cancelling_parent_cancels_and_drains_child(timeout: bool) -> None:
+    app: Final = FastAPI()
+    started: Final = asyncio.Event()
+    stopped: Final = asyncio.Event()
+
+    @app.post("/v1/chat/completions")
+    async def endpoint() -> None:
+        started.set()
+        try:
+            await asyncio.Event().wait()
+        finally:
+            stopped.set()
+
+    parent: Final = asyncio.create_task(with_proxy_compaction_executor(_child(), _request(app)))
+    await asyncio.wait_for(started.wait(), timeout=5)
+    if timeout:
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(parent, timeout=0)
+    else:
+        parent.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await parent
+    assert stopped.is_set() and compaction_executor.get() is None

tests/test_litellm/router_strategy/complexity_router/test_context_compaction.py

@@ -0,0 +1,503 @@
+import asyncio
+import json
+from collections.abc import Iterator, Mapping
+from copy import deepcopy
+from functools import partial
+from typing import Final, Literal
+
+import httpx
+import pytest
+import respx
+
+import litellm
+from litellm.llms import compaction as native
+from litellm.router_strategy.complexity_router.config import ContextCompactionConfig
+from litellm.router_strategy.complexity_router.context_compaction import (
+    CompactionState,
+    Surface,
+    arm_compaction,
+    compact_to_fit,
+    compaction_executor,
+)
+from litellm.types.llms.anthropic import ANTHROPIC_BETA_HEADER_VALUES
+from litellm.types.router import Deployment
+
+pytestmark: Final = [pytest.mark.asyncio, pytest.mark.usefixtures("local_model_cost_map")]
+SCHEMA: Final = {"type": "object", "properties": {"code": {"type": "string"}}}
+
+
+@pytest.fixture(autouse=True)
+def native_catalog(monkeypatch: pytest.MonkeyPatch, local_model_cost_map: None) -> None:
+    monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
+    monkeypatch.setenv("LITELLM_LICENSE", "")
+    monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", False)
+    monkeypatch.setitem(litellm.model_cost, "summary-fixture", {
+        "litellm_provider": "anthropic", "mode": "chat", "max_input_tokens": 32000,
+        "max_output_tokens": 4096, "supports_anthropic_compaction": True,
+    })
+
+
+def make_router(
+    window: int | None = 512, settings: Mapping[str, object] | None = None, *, compactor_window: int = 32000,
+    conflict: bool = False, output: int | None = 64,
+    answer_defaults: Mapping[str, object] | None = None,
+    context_fallback: bool = False,
+) -> litellm.Router:
+    config: Final = {
+        "tiers": {"SIMPLE": "small", "MEDIUM": "large", "COMPLEX": "large", "REASONING": "large"},
+        "keyword_tier_rules": [{"keywords": ["answer", "tail result"], "tier": "SIMPLE"}],
+        "enable_context_window_escalation": False, "max_tokens_from_tier_model": False,
+        **(settings or {}),
+    }
+    return litellm.Router(model_list=[
+        {"model_name": "auto", "litellm_params": {
+            "model": "auto_router/complexity_router", "complexity_router_config": config,
+        }},
+        {"model_name": "small", "litellm_params": {
+            "model": "openai/arbitrary-answer", "api_base": "https://answer.test/v1", "api_key": "answer-test", "max_retries": 0,
+            **(answer_defaults or {}),
+        }, "model_info": {"id": "pinned-answer", "max_input_tokens": window, "max_output_tokens": output}},
+        {"model_name": "large", "litellm_params": {
+            "model": "anthropic/summary-fixture", "api_base": "https://compact.test", "api_key": "compact-test",
+            **({"stop": ["deployment policy"]} if conflict else {}),
+        }, "model_info": {"id": "native-compactor", "max_input_tokens": compactor_window, "max_output_tokens": 4096}},
+        {"model_name": "backup", "litellm_params": {
+            "model": "anthropic/summary-fixture", "api_base": "https://compact.test", "api_key": "backup-test",
+        }, "model_info": {"id": "backup-compactor", "max_input_tokens": 32000, "max_output_tokens": 4096}},
+    ], enable_pre_call_checks=True, num_retries=0, disable_cooldowns=True,
+        retry_policy={"InternalServerErrorRetries": 1},
+        context_window_fallbacks=[{"auto": ["large"]}] if context_fallback else [])
+
+
+def exchange(surface: Surface, phase: str) -> list[dict[str, object]]:
+    identifier: Final = f"{phase}-call"
+    result: Final = f"{phase} result"
+    if surface == "responses":
+        return [
+            {"type": "function_call", "call_id": identifier, "name": "lookup", "arguments": "{}"},
+            {"type": "function_call_output", "call_id": identifier, "output": result},
+        ]
+    if surface == "messages":
+        return [
+            {"role": "assistant", "content": [{"type": "tool_use", "id": identifier, "name": "lookup", "input": {}}]},
+            {"role": "user", "content": [{"type": "tool_result", "tool_use_id": identifier, "content": result}]},
+        ]
+    return [
+        {"role": "assistant", "tool_calls": [
+            {"id": identifier, "type": "function", "function": {"name": "lookup", "arguments": "{}"}},
+        ]},
+        {"role": "tool", "tool_call_id": identifier, "content": result},
+    ]
+
+
+def history(surface: Surface) -> dict[str, object]:
+    conversation: Final = [
+        {"role": "user", "content": "Project code MAPLE-47. Background detail. " * 150},
+        {"role": "assistant", "content": "Recorded"},
+        *exchange(surface, "prefix"),
+        {"role": "user", "content": "Answer with the project code"},
+        *exchange(surface, "tail"),
+    ]
+    function: Final = {"name": "lookup", "parameters": SCHEMA}
+    if surface == "responses":
+        return {"instructions": "Keep the code exact", "tools": [{"type": "function", **function}], "input": [
+            {"role": "developer", "content": "Retain the original spelling"}, *conversation,
+        ]}
+    if surface == "messages":
+        return {"system": "Keep the code exact", "tools": [{"name": "lookup", "input_schema": SCHEMA}], "messages": [
+            *conversation,
+        ]}
+    return {"tools": [{"type": "function", "function": function}], "messages": [
+        {"role": "system", "content": "Keep the code exact"}, *conversation,
+    ]}
+
+
+def native_reply(summary: str = "Project code MAPLE-47", signed: bool = True, truncated: bool = False) -> httpx.Response:
+    return httpx.Response(200, json={
+        "id": "msg_compact", "type": "message", "role": "assistant", "model": "summary-fixture",
+        "content": [{"type": "compaction", "content": summary, **({"signature": "native-signature"} if signed else {})}],
+        "stop_reason": "max_tokens" if truncated else "compaction", "usage": {"input_tokens": 0, "output_tokens": 0, "iterations": [
+            {"type": "compaction", "input_tokens": 1200, "output_tokens": 20},
+        ]},
+    })
+
+
+def answer_reply(request: httpx.Request, expected_model: str = "arbitrary-answer") -> httpx.Response:
+    payload: Final = json.loads(request.content)
+    assert payload["model"] == expected_model
+    assert request.headers["authorization"] == "Bearer answer-test"
+    if request.url.path.endswith("responses"):
+        return httpx.Response(200, json={
+            "id": "resp_answer", "object": "response", "created_at": 0, "status": "completed",
+            "model": payload["model"], "output": [{"id": "msg_answer", "type": "message", "role": "assistant",
+                "status": "completed", "content": [{"type": "output_text", "text": "MAPLE-47", "annotations": []}]}],
+            "usage": {"input_tokens": 60, "output_tokens": 8, "total_tokens": 68},
+        })
+    return httpx.Response(200, json={
+        "id": "answer", "object": "chat.completion", "created": 0, "model": payload["model"],
+        "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "MAPLE-47"}}],
+        "usage": {"prompt_tokens": 60, "completion_tokens": 8, "total_tokens": 68},
+    })
+
+
+@pytest.fixture
+def wire() -> Iterator[tuple[respx.Route, respx.Route]]:
+    with respx.mock(assert_all_called=False) as transport:
+        compactor: Final = transport.post("https://compact.test/v1/messages").mock(return_value=native_reply())
+        answer: Final = transport.route(method="POST", host="answer.test").mock(side_effect=answer_reply)
+        yield compactor, answer
+
+
+async def invoke(router: litellm.Router, surface: Surface, payload: Mapping[str, object], retries: int = 0) -> object:
+    if surface == "responses":
+        return await router.aresponses(model="auto", max_output_tokens=64, num_retries=retries, **payload)
+    if surface == "messages":
+        return await router.aanthropic_messages(model="auto", max_tokens=64, num_retries=retries, **payload)
+    return await router.acompletion(model="auto", max_tokens=64, num_retries=retries, **payload)
+
+
+@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
+@pytest.mark.parametrize("near", [False, True])
+@pytest.mark.parametrize("configured", [False, True])
+async def test_all_surfaces_compact_and_keep_selected_answerer(
+    wire: tuple[respx.Route, respx.Route], surface: Surface, near: bool, configured: bool,
+) -> None:
+    payload: Final = history(surface)
+    original: Final = deepcopy(payload)
+    counted: Final = make_router()._count_pre_call_check_tokens(payload.get("messages"), payload.get("input"), payload)
+    window: Final = int((counted + 32) / ContextCompactionConfig().trigger_ratio) + 1 if near else 512
+    assert (counted < window) is near
+    settings: Final = {"enable_context_window_escalation": True,
+        **({"context_compaction": {"model": "large", "max_tokens": 512}} if configured else {})}
+    router: Final = make_router(window, settings)
+    captured: Final = asyncio.Queue[Mapping[str, object]]()
+    compactor, answer = wire
+    retry: Final = near and configured
+
+    def answer_after_retry(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, json={"error": {"message": "retry answer"}}) if answer.call_count == 0 else answer_reply(request)
+
+    if retry:
+        answer.mock(side_effect=answer_after_retry)
+
+    async def execute(
+        protocol: native.CompactionProtocol, request: Mapping[str, object], parent_model: str | None = None
+    ) -> Mapping[str, object]:
+        assert parent_model == "auto"
+        result: Final = await native.dispatch(router, protocol, request)
+        captured.put_nowait(result)
+        return result
+
+    token: Final = compaction_executor.set(execute)
+    try:
+        response: Final = await invoke(router, surface, payload, retries=int(retry))
+    finally:
+        compaction_executor.reset(token)
+    assert compactor.call_count == captured.qsize() == 1
+    assert answer.call_count == router.total_calls["openai/arbitrary-answer"] == 1 + int(retry)
+    compact_request: Final = compactor.calls[0].request
+    compact_body: Final = json.loads(compact_request.content)
+    answer_body: Final = json.loads(answer.calls[0].request.content)
+    assert answer_body == json.loads(answer.calls[-1].request.content)
+    assert compact_body["model"] == "summary-fixture" and compact_body["compaction"] == {"type": "summarize"}
+    assert ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_09_04.value in compact_request.headers["anthropic-beta"].split(",")
+    assert compact_body["max_tokens"] == (512 if configured else ContextCompactionConfig().max_tokens)
+    assert "Background detail" in str(compact_body) and "tail-call" not in str(compact_body)
+    assert "prefix-call" in str(compact_body) and "prefix result" in str(compact_body)
+    assert "Keep the code exact" in str(compact_body["system"])
+    assert compact_body["tools"][0]["input_schema"] == SCHEMA
+    assert "MAPLE-47" in str(answer_body) and "Background detail" not in str(answer_body)
+    assert "prefix-call" not in str(answer_body)
+    assert "native-signature" not in str(answer_body) and "compaction" not in answer_body
+    assert "tail-call" in str(answer_body) and "tail result" in str(answer_body)
+    assert "MAPLE-47" in str(response)
+    usage: Final = captured.get_nowait()["usage"]
+    if surface == "messages":
+        assert usage["iterations"][0]["input_tokens"] == 1200 and usage["iterations"][0]["output_tokens"] == 20
+    else:
+        assert usage["prompt_tokens"] == 1200 and usage["completion_tokens"] == 20
+    if surface == "responses":
+        assert answer_body["input"][-3:] == original["input"][-3:]
+        assert answer_body["input"][0] == original["input"][0]
+        assert answer_body["instructions"] == original["instructions"] and answer_body["tools"] == original["tools"]
+    assert payload == original
+
+
+@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
+@pytest.mark.parametrize("mode", ["fitting", "false", "null"])
+async def test_fitting_and_disabled_requests_do_not_compact(
+    wire: tuple[respx.Route, respx.Route], surface: Surface, mode: str,
+) -> None:
+    settings: Final = {} if mode == "fitting" else {"context_compaction": False if mode == "false" else None}
+    payload: Final = history(surface)
+    original: Final = deepcopy(payload)
+    counted: Final = make_router()._count_pre_call_check_tokens(payload.get("messages"), payload.get("input"), payload)
+    await invoke(make_router(20000 if mode == "fitting" else counted + 64, settings), surface, payload)
+    compactor, answer = wire
+    assert compactor.call_count == 0 and answer.call_count == 1
+    assert "Background detail" in answer.calls[0].request.content.decode()
+    assert payload == original
+
+
+@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
+@pytest.mark.parametrize("reason", ["single", "unclosed", "no_compactor"])
+async def test_fitting_request_survives_unavailable_compaction(
+    wire: tuple[respx.Route, respx.Route], surface: Surface, reason: str,
+) -> None:
+    key: Final = "input" if surface == "responses" else "messages"
+    items: Final = [{"role": "user", "content": "Answer with MAPLE-47. Detail. " * 80}]
+    payload: Final = history(surface) if reason == "no_compactor" else {key: (
+        items if reason == "single" else [*items, *exchange(surface, "open")[:1], {"role": "user", "content": "Answer"}]
+    )}
+    counted: Final = make_router()._count_pre_call_check_tokens(payload.get("messages"), payload.get("input"), payload)
+    settings: Final = {"tiers": {"SIMPLE": "small", "MEDIUM": "small", "COMPLEX": "small", "REASONING": "small"}} if reason == "no_compactor" else {}
+    await invoke(make_router(counted + 1, settings), surface, payload)
+    compactor, answer = wire
+    assert compactor.call_count == 0 and answer.call_count == 1
+    assert "Detail" in str(answer.calls[0].request.content) or "Background detail" in str(answer.calls[0].request.content)
+
+
+@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
+async def test_uncompactable_overflow_uses_explicit_context_fallback(
+    wire: tuple[respx.Route, respx.Route], surface: Surface,
+) -> None:
+    compactor, answer = wire
+    compactor.mock(return_value=httpx.Response(200, json={**native_reply().json(),
+        "content": [{"type": "text", "text": "MAPLE-47"}], "stop_reason": "end_turn"}))
+    payload: Final = {"input" if surface == "responses" else "messages": [
+        {"role": "user", "content": "Answer with MAPLE-47. Detail. " * 150},
+    ]}
+    await invoke(make_router(context_fallback=True), surface, payload)
+    assert answer.call_count == 0 and compactor.call_count == 1
+    assert "compaction" not in json.loads(compactor.calls[0].request.content)
+
+
+@pytest.mark.parametrize("escalate", [False, True])
+async def test_no_native_compactor_respects_explicit_escalation(
+    wire: tuple[respx.Route, respx.Route], monkeypatch: pytest.MonkeyPatch, escalate: bool,
+) -> None:
+    monkeypatch.setitem(litellm.model_cost["summary-fixture"], "supports_anthropic_compaction", False)
+    router: Final = make_router(settings={"enable_context_window_escalation": escalate})
+    compactor, answer = wire
+    compactor.mock(return_value=httpx.Response(200, json={**native_reply().json(),
+        "content": [{"type": "text", "text": "MAPLE-47"}], "stop_reason": "end_turn"}))
+    if not escalate:
+        with pytest.raises(litellm.ContextWindowExceededError, match="No configured compactor"):
+            await invoke(router, "chat", history("chat"))
+        assert compactor.call_count == 0
+    else:
+        await invoke(router, "chat", history("chat"))
+        assert compactor.call_count == 1
+        assert "compaction" not in json.loads(compactor.calls[0].request.content)
+    assert answer.call_count == 0
+
+
+@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
+async def test_undersized_native_compactor_does_not_block_explicit_escalation(
+    wire: tuple[respx.Route, respx.Route], surface: Surface,
+) -> None:
+    router: Final = make_router(compactor_window=512, settings={
+        "enable_context_window_escalation": True,
+        "tiers": {"SIMPLE": "small", "MEDIUM": "large", "COMPLEX": "wide", "REASONING": "wide"},
+    })
+    router.add_deployment(Deployment(
+        model_name="wide", litellm_params={
+            "model": "openai/wide-answer", "api_base": "https://answer.test/v1", "api_key": "answer-test",
+        }, model_info={"id": "wide-answer", "max_input_tokens": 32000, "max_output_tokens": 64},
+    ))
+    compactor, answer = wire
+    answer.mock(side_effect=partial(answer_reply, expected_model="wide-answer"))
+    await invoke(router, surface, history(surface))
+    assert compactor.call_count == 0 and answer.call_count == 1
+    assert json.loads(answer.calls[0].request.content)["model"] == "wide-answer"
+
+
+@pytest.mark.parametrize(
+    ("surface", "failure"),
+    [(surface, failure) for surface in ("chat", "messages", "responses") for failure in ("unsigned", "oversized", "provider")]
+    + [("messages", "truncated")],
+)
+async def test_bad_native_result_never_reaches_answerer(
+    wire: tuple[respx.Route, respx.Route], surface: Surface, failure: str,
+) -> None:
+    compactor, answer = wire
+    reply: Final = httpx.Response(500, json={"error": {"type": "api_error", "message": "failed"}}) if failure == "provider" else native_reply(
+        "too large " * 2000 if failure == "oversized" else "MAPLE-47", signed=failure != "unsigned",
+        truncated=failure == "truncated",
+    )
+    compactor.mock(return_value=reply)
+    with pytest.raises((litellm.BadRequestError, litellm.InternalServerError)):
+        await invoke(make_router(), surface, history(surface))
+    assert compactor.call_count == 1 and answer.call_count == 0
+
+
+@pytest.mark.parametrize("failure", ["item", "content", "tools", "unclosed", "missing", "duplicate", "instructions", "retained"])
+@pytest.mark.parametrize("needed", [False, True])
+async def test_unsafe_responses_reject_only_when_compaction_needed(
+    wire: tuple[respx.Route, respx.Route], failure: str, needed: bool,
+) -> None:
+    payload: Final = history("responses")
+    extra: Final = {
+        "item": [{"type": "computer_call", "call_id": "opaque-tool"}],
+        "content": [{"role": "assistant", "content": [{"type": "refusal", "refusal": "cannot"}]}],
+        "unclosed": [{"type": "function_call", "call_id": "unclosed", "name": "lookup", "arguments": "{}"}],
+        "missing": [{"type": "function_call", "name": "lookup", "arguments": "{}"}],
+        "duplicate": exchange("responses", "prefix"),
+        "instructions": [{"role": "developer", "content": "Changed instructions"}],
+    }
+    request: Final = {
+        **payload, "input": [*payload["input"][:3], *extra.get(failure, []), *payload["input"][3:]],
+        **({"tools": [{"type": "computer_use_preview", "display_width": 800, "display_height": 600}]} if failure == "tools" else {}),
+        **({"instructions": "Keep every instruction " * 600} if failure == "retained" else {}),
+    }
+    if needed:
+        with pytest.raises(litellm.BadRequestError, match="Context compaction"):
+            await invoke(make_router(), "responses", request)
+        assert all(route.call_count == 0 for route in wire)
+    else:
+        await invoke(make_router(20000), "responses", request)
+        compactor, answer = wire
+        assert compactor.call_count == 0 and answer.call_count == 1
+
+
+@pytest.mark.parametrize("owned", [
+    {"previous_response_id": "resp_parent"}, {"conversation": "conv_parent"},
+    {"context_management": [{"type": "compaction", "compact_threshold": 1000}]}, {"compaction": {"type": "summarize"}},
+    {"input": [{"type": "reasoning", "encrypted_content": "opaque"}]},
+    {"input": [{"type": "reasoning", "summary": [{"type": "summary_text", "text": "prior reasoning"}]}]},
+    {"input": [{"type": "compaction", "encrypted_content": "opaque"}]},
+    {"input": [{"type": "item_reference", "id": "item_parent"}]},
+    {"input": [{"role": "assistant", "content": "visible", "encrypted_content": "opaque"}]},
+    {"input": [{"role": "user", "content": [{"type": "encrypted_content", "encrypted_content": "opaque"}]}]},
+])
+@pytest.mark.parametrize("arm_first", [False, True])
+async def test_client_owned_history_bypasses_compaction(
+    wire: tuple[respx.Route, respx.Route], owned: Mapping[str, object], arm_first: bool,
+) -> None:
+    router: Final = make_router()
+    deployment: Final = router.get_deployment(model_id="pinned-answer")
+    assert deployment is not None
+    state: Final = CompactionState()
+    request: Final = {**history("responses"), **owned, "model": "small", "max_tokens": 64, "_context_compaction_state": state}
+    original: Final = deepcopy({key: value for key, value in request.items() if key != "_context_compaction_state"})
+    before_defaults: Final = {"_context_compaction_state": state} if arm_first else request
+    await arm_compaction(before_defaults, ContextCompactionConfig(), ("large",))
+    counted: Final = router._count_pre_call_check_tokens(None, request["input"], request)
+    if arm_first and counted > 512:
+        with pytest.raises(litellm.ContextWindowExceededError):
+            await compact_to_fit(router, deployment.model_dump(), request, "responses")
+    else:
+        result: Final = await compact_to_fit(router, deployment.model_dump(), request, "responses")
+        assert result is request
+    assert {key: value for key, value in request.items() if key != "_context_compaction_state"} == original
+    assert all(route.call_count == 0 for route in wire)
+
+
+@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
+@pytest.mark.parametrize("source", ["request", "deployment"])
+async def test_client_managed_overflow_keeps_context_window_admission(
+    wire: tuple[respx.Route, respx.Route], surface: Surface, source: str,
+) -> None:
+    managed: Final = {"context_management": {"edits": []}}
+    router: Final = make_router(answer_defaults=managed if source == "deployment" else None)
+    payload: Final = {**history(surface), **(managed if source == "request" else {})}
+    compactor, answer = wire
+    if source == "deployment":
+        with pytest.raises(litellm.ContextWindowExceededError):
+            await invoke(router, surface, payload)
+        assert compactor.call_count == 0
+    else:
+        await invoke(router, surface, payload)
+        assert compactor.call_count == 1
+        assert "compaction" not in json.loads(compactor.calls[0].request.content)
+    assert answer.call_count == 0
+
+
+@pytest.mark.parametrize("case", ["conflicting_defaults", "small_window", "capability_false", "capability_missing"])
+async def test_automatic_compactor_skips_conflicts_and_requires_capacity_and_capability(
+    wire: tuple[respx.Route, respx.Route], monkeypatch: pytest.MonkeyPatch, case: str,
+) -> None:
+    if case.startswith("capability"):
+        metadata: Final = {key: value for key, value in litellm.model_cost["summary-fixture"].items()
+                           if key != "supports_anthropic_compaction"}
+        monkeypatch.setitem(litellm.model_cost, "summary-fixture", {
+            **metadata, **({"supports_anthropic_compaction": False} if case == "capability_false" else {}),
+        })
+    conflict: Final = case == "conflicting_defaults"
+    settings: Final = {"tiers": {"SIMPLE": "small", "MEDIUM": "large", "COMPLEX": "backup", "REASONING": "backup"}} if conflict else {}
+    router: Final = make_router(settings=settings, conflict=conflict, compactor_window=512 if case == "small_window" else 32000)
+    if conflict:
+        await invoke(router, "chat", history("chat"))
+        compactor, answer = wire
+        assert compactor.call_count == answer.call_count == 1
+        assert compactor.calls[0].request.headers["x-api-key"] == "backup-test"
+        assert "stop_sequences" not in json.loads(compactor.calls[0].request.content)
+    else:
+        with pytest.raises(litellm.BadRequestError, match="No configured compactor"):
+            await invoke(router, "chat", history("chat"))
+        assert all(route.call_count == 0 for route in wire)
+
+
+@pytest.mark.parametrize("unknown_output", [False, True])
+@pytest.mark.parametrize("overflow", [False, True])
+async def test_unusable_output_budget_still_enforces_known_input_window(
+    wire: tuple[respx.Route, respx.Route], unknown_output: bool, overflow: bool,
+) -> None:
+    payload: Final = history("chat")
+    counted: Final = make_router(output=None)._count_pre_call_check_tokens(payload["messages"], None, payload)
+    window: Final = 512 if overflow else counted + 64
+    output: Final = None if unknown_output else window
+    router: Final = make_router(window, output=output)
+    deployment: Final = router.get_deployment(model_id="pinned-answer")
+    assert deployment is not None
+    state: Final = CompactionState(config=ContextCompactionConfig(), candidates=("large",))
+    request: Final = {**payload, "model": "small", "max_tokens": output, "_context_compaction_state": state}
+    if overflow:
+        with pytest.raises(litellm.BadRequestError, match="known input window"):
+            await compact_to_fit(router, deployment.model_dump(), request, "chat")
+    else:
+        assert await compact_to_fit(router, deployment.model_dump(), request, "chat") is request
+    assert all(route.call_count == 0 for route in wire)
+
+
+@pytest.mark.parametrize("outcome", ["success", "timeout", "cancel"])
+async def test_retry_reuses_summary_or_terminal_cancellation(outcome: Literal["success", "timeout", "cancel"]) -> None:
+    router: Final = make_router()
+    deployment: Final = router.get_deployment(model_id="pinned-answer")
+    assert deployment is not None
+    state: Final = CompactionState(config=ContextCompactionConfig(model="large", max_tokens=512, timeout_seconds=0.02))
+    request: Final = {**history("messages"), "model": "small", "max_tokens": 64, "_context_compaction_state": state}
+    calls: Final = asyncio.Queue[None]()
+    started: Final = asyncio.Event()
+    stopped: Final = asyncio.Event()
+
+    async def execute(
+        protocol: native.CompactionProtocol, payload: Mapping[str, object], parent_model: str | None = None
+    ) -> Mapping[str, object]:
+        calls.put_nowait(None)
+        started.set()
+        try:
+            return native_reply().json() if outcome == "success" else await asyncio.Future[Mapping[str, object]]()
+        finally:
+            stopped.set()
+
+    token: Final = compaction_executor.set(execute)
+    try:
+        first: Final = asyncio.create_task(compact_to_fit(router, deployment.model_dump(), request, "messages"))
+        await asyncio.wait_for(started.wait(), timeout=2)
+        if outcome == "cancel":
+            first.cancel()
+        if outcome == "success":
+            assert await first == await compact_to_fit(router, deployment.model_dump(), request, "messages")
+            changed: Final = {**request, "messages": [{"role": "user", "content": "new history"}, *request["messages"]]}
+            with pytest.raises(litellm.BadRequestError, match="History changed"):
+                await compact_to_fit(router, deployment.model_dump(), changed, "messages")
+        else:
+            error: Final = asyncio.CancelledError if outcome == "cancel" else asyncio.TimeoutError
+            with pytest.raises(error):
+                await first
+            with pytest.raises(error):
+                await compact_to_fit(router, deployment.model_dump(), request, "messages")
+        assert calls.qsize() == 1 and stopped.is_set()
+    finally:
+        compaction_executor.reset(token)

tests/test_litellm/router_strategy/test_complexity_router.py

@@ -15317,6 +15317,7 @@ class TestHealthFallbackDispatch:
 
         router: Final = self._router(
             config={
+                "context_compaction": False,
                 "tiers": {"SIMPLE": "primary", "MEDIUM": "peer", "COMPLEX": "large"},
                 "enable_context_window_escalation": True,
             }
@@ -15399,6 +15400,7 @@ class TestHealthFallbackDispatch:
     async def test_modality_default_must_also_fit_context(self, default_fits: bool) -> None:
         router: Final = self._router(
             config={
+                "context_compaction": False,
                 "modality_routing": True,
                 "tiers": {"SIMPLE": "primary"},
                 "enable_context_window_escalation": True,

tests/test_litellm/test_anthropic_beta_headers_filtering.py

@@ -83,6 +83,7 @@ class TestAnthropicBetaHeadersFiltering:
         filtered = filter_and_transform_beta_headers(
             beta_headers=all_headers, provider=provider
         )
+        assert ("compact-2026-09-04" in filtered) is (provider == "anthropic")
 
         for header in unsupported_headers:
             assert (

tests/test_litellm/test_utils.py

@@ -178,6 +178,14 @@ def test_potential_model_names_keeps_provider_prefixed_candidate():
     assert bare["provider_prefixed_model_name"] == bare["combined_model_name"] == "perplexity/glm-5.2"
 
 
+@pytest.mark.parametrize("capability", [True, False, None])
+def test_get_model_info_anthropic_compaction(
+    local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch, capability: bool | None
+) -> None:
+    monkeypatch.setitem(litellm.model_cost["claude-sonnet-5"], "supports_anthropic_compaction", capability)
+    assert litellm.get_model_info("claude-sonnet-5")["supports_anthropic_compaction"] is capability
+
+
 def test_get_model_info_strips_openai_finetune_ids_without_a_custom_suffix(local_model_cost_map):
     info = litellm.get_model_info(model="ft:gpt-4o-2024-08-06:my-org::abc123", custom_llm_provider="openai")
     assert info["key"] == "ft:gpt-4o-2024-08-06"
@@ -861,6 +869,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
                 "source": {"type": "string"},
                 "comment": {"type": "string"},
                 "supports_assistant_prefill": {"type": "boolean"},
+                "supports_anthropic_compaction": {"type": "boolean"},
                 "supports_audio_input": {"type": "boolean"},
                 "supports_audio_output": {"type": "boolean"},
                 "gemini_native_audio": {"type": "boolean"},

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -27678,6 +27678,26 @@ export interface components {
              */
             pattern_type: "prebuilt" | "regex";
         };
+        /** ContextCompactionConfig */
+        ContextCompactionConfig: {
+            /**
+             * Max Tokens
+             * @default 4096
+             */
+            max_tokens: number;
+            /** Model */
+            model?: string | null;
+            /**
+             * Timeout Seconds
+             * @default 120
+             */
+            timeout_seconds: number;
+            /**
+             * Trigger Ratio
+             * @default 0.9
+             */
+            trigger_ratio: number;
+        };
         /**
          * CoordinationRedisNode
          * @description A single startup node of a cluster-mode Redis used for proxy coordination.
@@ -36962,6 +36982,11 @@ export interface components {
              * @description Keywords indicating code-related content
              */
             code_keywords?: string[] | null;
+            /**
+             * Context Compaction
+             * @description Compact full conversation history near the selected deployment's input limit for Chat, Responses and Messages. Uses a capable configured tier model unless model is specified. Set false or null to disable. Stored and client-managed native history keep their existing behavior.
+             */
+            context_compaction?: components["schemas"]["ContextCompactionConfig"] | false;
             /**
              * Context Window Escalation Buffer
              * @description Fraction of a model's declared context window the estimated prompt must fit within. The token count is an estimate, so fitting against the full window would dispatch prompts that the provider's own tokenizer then rejects; 0.95 leaves room for that drift plus the response tokens.