diff --git a/studio/backend/core/inference/external_provider.py b/studio/backend/core/inference/external_provider.py
index 71a5293f7e..28027841cf 100644
--- a/studio/backend/core/inference/external_provider.py
+++ b/studio/backend/core/inference/external_provider.py
@@ -124,6 +124,29 @@ def _anthropic_code_execution_version(model: str) -> str:
_ANTHROPIC_CODE_EXECUTION_BETA = "code-execution-2025-08-25"
+# Anthropic server-side context compaction (beta as of compact-2026-01-12).
+# Per the docs, the compaction tool is currently supported on Opus 4.6,
+# Opus 4.7, Sonnet 4.6 and Mythos Preview. The beta header is the same
+# for every supported model; the dated `compact_20260112` type lives in
+# the body's `context_management.edits` array. Anything sent to a model
+# outside this prefix list is silently ignored so we don't 400 upstream.
+_ANTHROPIC_COMPACTION_PREFIXES = (
+ "claude-opus-4-7",
+ "claude-opus-4-6",
+ "claude-sonnet-4-6",
+ "claude-mythos-preview",
+)
+_ANTHROPIC_COMPACTION_BETA = "compact-2026-01-12"
+_ANTHROPIC_COMPACTION_TYPE = "compact_20260112"
+# The docs require the threshold to be at least 50K tokens; lower values
+# would 400. We clamp on the way out so a UI slider can't underflow.
+_ANTHROPIC_COMPACTION_MIN = 50_000
+
+
+def _anthropic_supports_compaction(model: str) -> bool:
+ return model.startswith(_ANTHROPIC_COMPACTION_PREFIXES)
+
+
class _MistralThinkingSpec(NamedTuple):
models: tuple[str, ...]
style: Literal["prompt_mode", "reasoning_effort", "disabled"]
@@ -294,6 +317,7 @@ class ExternalProviderClient:
openai_code_exec_container_id: Optional[str] = None,
anthropic_code_exec_container_id: Optional[str] = None,
prompt_cache_ttl: Optional[str] = None,
+ compaction_threshold: Optional[int] = None,
stream: bool = True,
) -> AsyncGenerator[str, None]:
"""
@@ -321,6 +345,7 @@ class ExternalProviderClient:
enable_prompt_caching,
anthropic_code_exec_container_id,
prompt_cache_ttl,
+ compaction_threshold,
):
yield line
return
@@ -1129,6 +1154,7 @@ class ExternalProviderClient:
enable_prompt_caching: Optional[bool] = None,
anthropic_code_exec_container_id: Optional[str] = None,
prompt_cache_ttl: Optional[str] = None,
+ compaction_threshold: Optional[int] = None,
) -> AsyncGenerator[str, None]:
"""
Call the Anthropic Messages API and translate its SSE to OpenAI format.
@@ -1163,6 +1189,20 @@ class ExternalProviderClient:
for part in content:
if part.get("type") == "text":
anthropic_parts.append({"type": "text", "text": part["text"]})
+ elif part.get("type") == "compaction":
+ # Round-trip the compaction block. When the
+ # prior assistant turn ran server-side
+ # compaction, that block must land back on this
+ # turn's assistant message so Anthropic skips
+ # re-compaction from scratch. Forward verbatim
+ # under the {type:"compaction", content:"..."}
+ # shape the API expects. See
+ # https://platform.claude.com/docs/en/build-with-claude/compaction
+ summary = part.get("content") or ""
+ if isinstance(summary, str) and summary:
+ anthropic_parts.append(
+ {"type": "compaction", "content": summary}
+ )
elif part.get("type") == "image_url":
url = part.get("image_url", {}).get("url", "")
if url.startswith("data:"):
@@ -1446,6 +1486,40 @@ class ExternalProviderClient:
if anthropic_code_exec_container_id:
body["container"] = anthropic_code_exec_container_id
+ # Server-side context compaction — see
+ # https://platform.claude.com/docs/en/build-with-claude/compaction
+ # Beta as of `compact-2026-01-12`. When `compaction_threshold` is
+ # provided AND the model accepts compaction (Opus 4.6+ / 4.7,
+ # Sonnet 4.6, Mythos preview), attach
+ # `context_management.edits[{type:"compact_20260112", trigger:
+ # {type:"input_tokens", value:N}}]` to the body. Anthropic runs
+ # the compaction step server-side once the rendered prompt
+ # crosses the threshold and replies with a top-level
+ # `context_management` block plus `usage.iterations[]` so we can
+ # account per-iteration. Below-min thresholds get clamped up to
+ # 50K so the request doesn't 400.
+ compaction_active = (
+ compaction_threshold is not None
+ and compaction_threshold > 0
+ and _anthropic_supports_compaction(model)
+ )
+ if compaction_active:
+ trigger_value = max(
+ int(compaction_threshold),
+ _ANTHROPIC_COMPACTION_MIN,
+ )
+ body["context_management"] = {
+ "edits": [
+ {
+ "type": _ANTHROPIC_COMPACTION_TYPE,
+ "trigger": {
+ "type": "input_tokens",
+ "value": trigger_value,
+ },
+ }
+ ]
+ }
+
url = f"{self.base_url}/messages"
completion_id = f"chatcmpl-anthropic-{model.replace('/', '-')}"
@@ -1489,20 +1563,22 @@ class ExternalProviderClient:
logger.info("Proxying Anthropic Messages API to %s (model=%s)", url, model)
request_headers = self._auth_headers()
- if code_execution_enabled:
- # Anthropic accepts comma-separated beta features in a single
- # `anthropic-beta` header. Merge our flag onto whatever the
- # registry's extra_headers contributed (currently nothing on
- # the beta axis, just anthropic-version) so future betas
- # added at the registry level keep working.
- existing_beta = request_headers.get("anthropic-beta", "").strip()
- beta_parts = (
- [p.strip() for p in existing_beta.split(",") if p.strip()]
- if existing_beta
- else []
- )
- if _ANTHROPIC_CODE_EXECUTION_BETA not in beta_parts:
- beta_parts.append(_ANTHROPIC_CODE_EXECUTION_BETA)
+ # Anthropic accepts comma-separated beta features in a single
+ # `anthropic-beta` header. Merge our flags onto whatever the
+ # registry's extra_headers contributed (currently nothing on
+ # the beta axis, just anthropic-version) so future betas
+ # added at the registry level keep working.
+ existing_beta = request_headers.get("anthropic-beta", "").strip()
+ beta_parts = (
+ [p.strip() for p in existing_beta.split(",") if p.strip()]
+ if existing_beta
+ else []
+ )
+ if code_execution_enabled and _ANTHROPIC_CODE_EXECUTION_BETA not in beta_parts:
+ beta_parts.append(_ANTHROPIC_CODE_EXECUTION_BETA)
+ if compaction_active and _ANTHROPIC_COMPACTION_BETA not in beta_parts:
+ beta_parts.append(_ANTHROPIC_COMPACTION_BETA)
+ if beta_parts:
request_headers["anthropic-beta"] = ",".join(beta_parts)
try:
@@ -1589,6 +1665,17 @@ class ExternalProviderClient:
current_web_fetch_use: Optional[dict[str, Any]] = None
current_web_fetch_result: Optional[dict[str, Any]] = None
web_fetch_calls: dict[str, dict[str, Any]] = {}
+ # Compaction state. Server-side compaction emits a
+ # `{type:"compaction", content:"..."}` content block
+ # whenever it runs. The summary text can land on the
+ # start event AND/OR via text_delta events on the same
+ # block (Anthropic's wire format is permissive here).
+ # Accumulate in `current_compaction["content"]` and emit
+ # on content_block_stop so the chat-adapter can persist
+ # it onto the assistant message for round-tripping on
+ # the next turn.
+ current_compaction: Optional[dict[str, Any]] = None
+ compaction_blocks_seen = 0
# Counts surfaced in the final log line so reports of
# "Code execution did nothing" can be triaged at a
# glance. generated_files_count is interesting for the
@@ -1884,6 +1971,23 @@ class ExternalProviderClient:
"tool_use_id": tool_use_id,
"inner": inner if isinstance(inner, dict) else {},
}
+ elif block_type == "compaction":
+ # Server-side compaction emits a `compaction`
+ # content block on the assistant message.
+ # Anthropic may include the summary text on
+ # this start event AND/OR stream it via
+ # text_delta events on the same block. See
+ # https://platform.claude.com/docs/en/build-with-claude/compaction
+ # Capture either form; finalize and emit
+ # on content_block_stop. The chat-adapter
+ # persists the block onto the assistant
+ # message so the next turn's request
+ # carries it back -- Anthropic then skips
+ # re-compaction from scratch.
+ seed = content_block.get("content") or ""
+ current_compaction = {
+ "content": seed if isinstance(seed, str) else "",
+ }
elif event_type == "content_block_delta":
delta = event.get("delta", {})
@@ -1902,21 +2006,31 @@ class ExternalProviderClient:
thinking_open = True
yield _content_chunk(thinking_text)
elif delta_type == "text_delta":
- # First text after a thinking block closes the
- # tag we opened above. Anthropic emits
- # a content_block_stop between blocks, but
- # closing on the text_delta transition is more
- # forgiving if events arrive out of order.
- if thinking_open:
- yield _content_chunk("")
- thinking_open = False
text = delta.get("text", "")
- if text:
- yield _content_chunk(text)
- # Citations on text deltas are attached
- # per-call by Anthropic via the
- # `web_search_tool_result` block; we don't
- # need to scrape them off the text events.
+ # text_deltas inside a compaction block
+ # carry the summary chunks; route them
+ # into the compaction buffer and DON'T
+ # yield them to the user-visible stream
+ # -- the summary is opaque internal
+ # state, not assistant prose.
+ if current_compaction is not None:
+ if text:
+ current_compaction["content"] += text
+ else:
+ # First text after a thinking block closes the
+ # tag we opened above. Anthropic emits
+ # a content_block_stop between blocks, but
+ # closing on the text_delta transition is more
+ # forgiving if events arrive out of order.
+ if thinking_open:
+ yield _content_chunk("")
+ thinking_open = False
+ if text:
+ yield _content_chunk(text)
+ # Citations on text deltas are attached
+ # per-call by Anthropic via the
+ # `web_search_tool_result` block; we don't
+ # need to scrape them off the text events.
elif delta_type == "input_json_delta":
# Streamed partial_json carrying tool inputs
# — the search query for web_search, or the
@@ -2019,6 +2133,23 @@ class ExternalProviderClient:
}
)
current_code_exec_use = None
+ elif current_compaction is not None:
+ # End of a compaction block. Emit it as a
+ # synthetic tool_event so the chat-adapter
+ # can persist the {type:"compaction",
+ # content:"..."} payload onto the
+ # assistant message. The next turn's
+ # request body forwards the content_part
+ # verbatim and Anthropic recognises it
+ # as the prior compaction state.
+ compaction_blocks_seen += 1
+ yield _emit_tool_event(
+ {
+ "type": "compaction_block",
+ "content": current_compaction["content"],
+ }
+ )
+ current_compaction = None
elif current_code_exec_result is not None:
# End of a code-execution result block —
# format the inner result into the text
@@ -2120,6 +2251,33 @@ class ExternalProviderClient:
delta_usage = event.get("usage")
if isinstance(delta_usage, dict):
last_usage.update(delta_usage)
+ # When a fresh compaction has run, Anthropic
+ # publishes per-iteration token counts in
+ # `usage.iterations[]`. The top-level
+ # input_tokens / output_tokens only cover the
+ # `message` iteration, NOT the compaction
+ # passes — billing has to sum the whole
+ # array. See
+ # https://platform.claude.com/docs/en/build-with-claude/compaction
+ # Fold the compaction iterations into
+ # `compaction_input_tokens` / `compaction_output_tokens`
+ # so the cost surface can add them without
+ # re-walking the array (and so the closing
+ # log line names the figures).
+ iterations = delta_usage.get("iterations")
+ if isinstance(iterations, list):
+ c_in = 0
+ c_out = 0
+ for it in iterations:
+ if (
+ isinstance(it, dict)
+ and it.get("type") == "compaction"
+ ):
+ c_in += int(it.get("input_tokens") or 0)
+ c_out += int(it.get("output_tokens") or 0)
+ if c_in or c_out:
+ last_usage["compaction_input_tokens"] = c_in
+ last_usage["compaction_output_tokens"] = c_out
# Anthropic reports the code_execution container
# id on `message_delta.delta.container.{id,
# expires_at}` (NOT on message_start — at start
@@ -2244,7 +2402,10 @@ class ExternalProviderClient:
"container_id_in=%s, container_id_out=%s, "
"input_tokens=%s, output_tokens=%s, "
"cache_creation_input_tokens=%s, "
- "cache_read_input_tokens=%s, events=%s)",
+ "cache_read_input_tokens=%s, "
+ "compaction_input_tokens=%s, "
+ "compaction_output_tokens=%s, "
+ "compaction_blocks_seen=%s, events=%s)",
model,
web_search_requested,
web_search_invocations,
@@ -2263,6 +2424,9 @@ class ExternalProviderClient:
last_usage.get("output_tokens"),
last_usage.get("cache_creation_input_tokens"),
last_usage.get("cache_read_input_tokens"),
+ last_usage.get("compaction_input_tokens"),
+ last_usage.get("compaction_output_tokens"),
+ compaction_blocks_seen,
event_counts,
)
await response.aclose()
diff --git a/studio/backend/models/inference.py b/studio/backend/models/inference.py
index b2376f7be4..69d6b40c91 100644
--- a/studio/backend/models/inference.py
+++ b/studio/backend/models/inference.py
@@ -440,6 +440,28 @@ class ImageContentPart(BaseModel):
image_url: ImageUrl
+class CompactionContentPart(BaseModel):
+ """Anthropic server-side compaction state, attached to an assistant
+ message for round-tripping on the next turn.
+
+ When Anthropic runs compaction during a request, the response
+ carries a ``{"type": "compaction", "content": ""}`` block
+ on the assistant message. The chat-adapter persists it onto the
+ stored message; the next turn's outbound request must forward it
+ back so Anthropic recognises the existing compaction state and
+ doesn't re-summarise the conversation from scratch. See
+ ``external_provider._stream_anthropic`` for the wire-side handling
+ and https://platform.claude.com/docs/en/build-with-claude/compaction
+ for the upstream contract.
+ """
+
+ type: Literal["compaction"]
+ content: str = Field(
+ ...,
+ description = "Anthropic-produced summary of the compacted-away conversation prefix.",
+ )
+
+
def _content_part_discriminator(v):
if isinstance(v, dict):
return v.get("type")
@@ -450,6 +472,7 @@ ContentPart = Annotated[
Union[
Annotated[TextContentPart, Tag("text")],
Annotated[ImageContentPart, Tag("image_url")],
+ Annotated[CompactionContentPart, Tag("compaction")],
],
Discriminator(_content_part_discriminator),
]
@@ -681,6 +704,23 @@ class ChatCompletionRequest(BaseModel):
"API 422 on the request. No-op on every non-Anthropic provider."
),
)
+ compaction_threshold: Optional[int] = Field(
+ None,
+ ge = 1,
+ le = 2_000_000,
+ description = (
+ "[x-unsloth] Anthropic server-side context compaction trigger, in "
+ "input tokens. When set on a compaction-capable model (Opus 4.6+, "
+ "Opus 4.7, Sonnet 4.6, Mythos preview), Studio attaches the "
+ "`compact_20260112` edit and the `compact-2026-01-12` beta header. "
+ "The minimum upstream-accepted threshold is 50k input tokens; "
+ "any value below that is clamped UP server-side in "
+ "`_stream_anthropic`. Kept permissive at the schema layer so the "
+ "in-helper clamp can run instead of returning 422 on a sub-50k "
+ "value the frontend may have stashed in localStorage. No-op on "
+ "any other provider or unsupported model."
+ ),
+ )
openai_code_exec_container_id: Optional[str] = Field(
None,
description = (
diff --git a/studio/backend/routes/inference.py b/studio/backend/routes/inference.py
index 8baf0b4198..efe99a942c 100644
--- a/studio/backend/routes/inference.py
+++ b/studio/backend/routes/inference.py
@@ -1677,13 +1677,21 @@ def _extract_content_parts(
def _build_external_messages(
messages: list,
supports_vision: bool,
+ provider_type: Optional[str] = None,
) -> list[dict]:
"""
Convert ChatMessage list to OpenAI-compatible dicts for external providers.
- - Vision providers: preserve multimodal content arrays (image_url parts intact).
- - Non-vision providers: flatten to text-only (images silently dropped).
+ Behaviour per content-part type:
+ - `text`: always preserved.
+ - `image_url`: preserved on vision providers; stripped on non-vision.
+ - `compaction`: Anthropic-only synthetic part (round-trips server-side
+ compaction state). Forwarded ONLY when provider_type=="anthropic";
+ stripped for every other provider so the unknown part doesn't
+ reach generic /chat/completions passthrough where it would 400
+ (e.g. DeepSeek, Mistral, Gemini, Kimi, OpenRouter, etc.).
"""
+ anthropic = provider_type == "anthropic"
result = []
for msg in messages:
if isinstance(msg.content, str):
@@ -1704,11 +1712,30 @@ def _build_external_messages(
"image_url": {"url": part.image_url.url},
}
)
+ elif part.type == "compaction" and anthropic:
+ # Anthropic stream helper forwards this as a
+ # native `compaction` block; every other
+ # provider would 400 on the unknown part, so
+ # gate by provider_type.
+ parts.append({"type": "compaction", "content": part.content})
result.append({"role": msg.role, "content": parts})
else:
- # Non-vision provider — strip images, keep text only
- text = "\n".join(p.text for p in msg.content if p.type == "text")
- result.append({"role": msg.role, "content": text})
+ # Non-vision provider: keep text, optionally keep
+ # compaction (Anthropic only -- compaction-capable
+ # Anthropic models all report supports_vision=True
+ # today, but the gate is here for safety).
+ preserved = []
+ for p in msg.content:
+ if p.type == "text":
+ preserved.append({"type": "text", "text": p.text})
+ elif p.type == "compaction" and anthropic:
+ preserved.append({"type": "compaction", "content": p.content})
+ if len(preserved) == 1 and preserved[0]["type"] == "text":
+ # Single text part collapses back to a string for
+ # providers that don't accept content arrays.
+ result.append({"role": msg.role, "content": preserved[0]["text"]})
+ else:
+ result.append({"role": msg.role, "content": preserved})
return result
@@ -1779,7 +1806,11 @@ async def _proxy_to_external_provider(
_pinfo = _get_provider_info(provider_type) or {}
_supports_vision = _pinfo.get("supports_vision", False)
- chat_messages = _build_external_messages(payload.messages, _supports_vision)
+ chat_messages = _build_external_messages(
+ payload.messages,
+ _supports_vision,
+ provider_type = provider_type,
+ )
client = ExternalProviderClient(
provider_type = provider_type,
@@ -1803,6 +1834,7 @@ async def _proxy_to_external_provider(
openai_code_exec_container_id = payload.openai_code_exec_container_id,
anthropic_code_exec_container_id = payload.anthropic_code_exec_container_id,
prompt_cache_ttl = payload.prompt_cache_ttl,
+ compaction_threshold = payload.compaction_threshold,
stream = payload.stream,
)
try:
diff --git a/studio/backend/tests/test_anthropic_compaction.py b/studio/backend/tests/test_anthropic_compaction.py
new file mode 100644
index 0000000000..92b9280146
--- /dev/null
+++ b/studio/backend/tests/test_anthropic_compaction.py
@@ -0,0 +1,648 @@
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
+
+"""Unit tests for Anthropic server-side context compaction wiring.
+
+Compaction is a beta feature (header ``compact-2026-01-12``) gated to
+Opus 4.6, Opus 4.7, Sonnet 4.6, and Mythos preview. When enabled,
+Studio attaches ``context_management.edits[{type:"compact_20260112",
+trigger:{type:"input_tokens", value:N}}]`` to the outbound body. The
+minimum upstream-accepted threshold is 50k tokens; lower values are
+clamped to 50k so the request doesn't 400.
+
+These tests pin: the body shape per model, the beta header merge with
+the existing code-execution beta, threshold clamping, and silent no-op
+on unsupported models.
+"""
+
+import asyncio
+import json
+
+import httpx
+import pytest
+
+from core.inference import external_provider as ep_mod
+from core.inference.external_provider import (
+ ExternalProviderClient,
+ _anthropic_supports_compaction,
+)
+
+
+def _drive(coro):
+ return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_client() -> ExternalProviderClient:
+ return ExternalProviderClient(
+ provider_type = "anthropic",
+ base_url = "https://api.anthropic.com/v1",
+ api_key = "sk-ant-test",
+ )
+
+
+def _capture(monkeypatch, model: str, threshold, tools = None) -> dict:
+ captured: dict = {}
+
+ def handler(request: httpx.Request) -> httpx.Response:
+ captured["body"] = json.loads(request.content.decode("utf-8"))
+ captured["headers"] = dict(request.headers)
+ return httpx.Response(
+ 200,
+ content = b'event: message_stop\ndata: {"type": "message_stop"}\n\n',
+ headers = {"content-type": "text/event-stream"},
+ )
+
+ monkeypatch.setattr(
+ ep_mod,
+ "_http_client",
+ httpx.AsyncClient(transport = httpx.MockTransport(handler)),
+ )
+
+ async def run():
+ client = _make_client()
+ async for _ in client.stream_chat_completion(
+ messages = [{"role": "user", "content": "hi"}],
+ model = model,
+ temperature = 0.7,
+ top_p = 0.95,
+ max_tokens = 32,
+ enabled_tools = tools,
+ compaction_threshold = threshold,
+ ):
+ pass
+ await client.close()
+
+ _drive(run())
+ return captured
+
+
+# ── support gate matches the doc table ───────────────────────────────
+
+
+@pytest.mark.parametrize(
+ "model, supported",
+ [
+ ("claude-opus-4-7", True),
+ ("claude-opus-4-6", True),
+ ("claude-sonnet-4-6", True),
+ ("claude-mythos-preview", True),
+ # NOT supported per the docs.
+ ("claude-opus-4-5-20251101", False),
+ ("claude-sonnet-4-5-20250929", False),
+ ("claude-haiku-4-5-20251001", False),
+ ("claude-opus-4-1-20250805", False),
+ ("claude-opus-4-20250514", False),
+ ("claude-sonnet-4-20250514", False),
+ ("claude-3-5-sonnet-20241022", False),
+ ],
+)
+def test_supports_compaction_gate(model, supported):
+ assert _anthropic_supports_compaction(model) is supported
+
+
+# ── outbound shape on supported model ────────────────────────────────
+
+
+def test_supported_model_attaches_compaction_block_and_beta(monkeypatch):
+ captured = _capture(monkeypatch, "claude-opus-4-7", 150_000)
+ cm = captured["body"].get("context_management")
+ assert cm == {
+ "edits": [
+ {
+ "type": "compact_20260112",
+ "trigger": {"type": "input_tokens", "value": 150_000},
+ }
+ ]
+ }, cm
+ assert "compact-2026-01-12" in captured["headers"].get("anthropic-beta", "")
+
+
+def test_threshold_clamped_to_50k_minimum(monkeypatch):
+ # Below-min values get clamped UP so we don't 400 upstream.
+ captured = _capture(monkeypatch, "claude-opus-4-7", 60_000)
+ assert (
+ captured["body"]["context_management"]["edits"][0]["trigger"]["value"] == 60_000
+ )
+ captured = _capture(monkeypatch, "claude-opus-4-7", 1)
+ assert (
+ captured["body"]["context_management"]["edits"][0]["trigger"]["value"] == 50_000
+ )
+
+
+# ── beta header merge with code execution ────────────────────────────
+
+
+def test_compaction_beta_merges_with_code_execution_beta(monkeypatch):
+ captured = _capture(
+ monkeypatch,
+ "claude-opus-4-7",
+ 150_000,
+ tools = ["code_execution"],
+ )
+ beta = captured["headers"].get("anthropic-beta", "")
+ assert "code-execution-2025-08-25" in beta
+ assert "compact-2026-01-12" in beta
+
+
+# ── silent no-op on unsupported model ────────────────────────────────
+
+
+def test_unsupported_model_silently_drops_compaction(monkeypatch):
+ captured = _capture(monkeypatch, "claude-haiku-4-5-20251001", 150_000)
+ assert "context_management" not in captured["body"]
+ # The beta header must not carry compact-2026-01-12 either.
+ assert "compact-2026-01-12" not in captured["headers"].get(
+ "anthropic-beta",
+ "",
+ )
+
+
+# ── omitted threshold leaves body untouched ─────────────────────────
+
+
+def test_omitted_threshold_no_body_field(monkeypatch):
+ captured = _capture(monkeypatch, "claude-opus-4-7", None)
+ assert "context_management" not in captured["body"]
+ assert "compact-2026-01-12" not in captured["headers"].get(
+ "anthropic-beta",
+ "",
+ )
+
+
+# ── ChatCompletionRequest schema accepts sub-50k threshold ──────────
+
+
+def test_chat_completion_request_accepts_sub_50k_compaction_threshold():
+ # Codex P1 caught that ge=50_000 on the field caused FastAPI to
+ # 422 the request before the in-helper clamp could fire. The
+ # schema must accept any positive int and let _stream_anthropic
+ # clamp upward.
+ from models.inference import ChatCompletionRequest
+
+ req = ChatCompletionRequest.model_validate(
+ {
+ "model": "default",
+ "messages": [{"role": "user", "content": "hi"}],
+ "compaction_threshold": 1,
+ }
+ )
+ assert req.compaction_threshold == 1
+
+ req = ChatCompletionRequest.model_validate(
+ {
+ "model": "default",
+ "messages": [{"role": "user", "content": "hi"}],
+ "compaction_threshold": 49_999,
+ }
+ )
+ assert req.compaction_threshold == 49_999
+
+ # Non-positive values are still rejected so blank-string posts
+ # don't sneak through.
+ with pytest.raises(Exception):
+ ChatCompletionRequest.model_validate(
+ {
+ "model": "default",
+ "messages": [{"role": "user", "content": "hi"}],
+ "compaction_threshold": 0,
+ }
+ )
+
+
+# ── usage.iterations[] surfaces compaction tokens ──────────────────
+
+
+def test_message_delta_iterations_array_aggregates_compaction_tokens(
+ monkeypatch, capsys
+):
+ # When Anthropic compacts mid-stream, the SSE message_delta usage
+ # payload carries `iterations: [{type:"compaction", ...}, ...]`.
+ # The top-level input_tokens / output_tokens only account for the
+ # `message` iteration, so the cost surface needs the compaction
+ # totals exposed separately. The stream helper folds them into
+ # last_usage as `compaction_input_tokens` / `compaction_output_tokens`
+ # and surfaces them in the closing summary log so an operator can
+ # eyeball "did compaction cost us 180k tokens this turn?".
+
+ def http_handler(request: httpx.Request) -> httpx.Response:
+ body = (
+ b"event: message_start\n"
+ b'data: {"type":"message_start","message":{"usage":{"input_tokens":23000,"output_tokens":0}}}\n\n'
+ b"event: message_delta\n"
+ b'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},'
+ b'"usage":{"input_tokens":23000,"output_tokens":1000,'
+ b'"iterations":['
+ b'{"type":"compaction","input_tokens":180000,"output_tokens":3500},'
+ b'{"type":"message","input_tokens":23000,"output_tokens":1000}'
+ b"]}}\n\n"
+ b"event: message_stop\n"
+ b'data: {"type":"message_stop"}\n\n'
+ )
+ return httpx.Response(
+ 200,
+ content = body,
+ headers = {"content-type": "text/event-stream"},
+ )
+
+ monkeypatch.setattr(
+ ep_mod,
+ "_http_client",
+ httpx.AsyncClient(transport = httpx.MockTransport(http_handler)),
+ )
+
+ async def run():
+ client = _make_client()
+ async for _ in client.stream_chat_completion(
+ messages = [{"role": "user", "content": "hi"}],
+ model = "claude-opus-4-7",
+ temperature = 0.7,
+ top_p = 0.95,
+ max_tokens = 32,
+ compaction_threshold = 150_000,
+ ):
+ pass
+ await client.close()
+
+ _drive(run())
+
+ # structlog renders the closing summary through the stdlib bridge,
+ # which lands on stdout. Capture and check the rendered line.
+ out = capsys.readouterr().out
+ summary = next(
+ (line for line in out.splitlines() if "Anthropic stream complete" in line),
+ "",
+ )
+ assert "compaction_input_tokens=180000" in summary, summary
+ assert "compaction_output_tokens=3500" in summary, summary
+
+
+def test_message_delta_no_iterations_leaves_compaction_keys_unset(monkeypatch, capsys):
+ # Re-applying a previous compaction block does NOT emit a fresh
+ # iterations array. The helper must not invent compaction keys
+ # in that case (would otherwise double-bill).
+ def http_handler(request: httpx.Request) -> httpx.Response:
+ body = (
+ b"event: message_delta\n"
+ b'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},'
+ b'"usage":{"input_tokens":1234,"output_tokens":5}}\n\n'
+ b"event: message_stop\n"
+ b'data: {"type":"message_stop"}\n\n'
+ )
+ return httpx.Response(
+ 200,
+ content = body,
+ headers = {"content-type": "text/event-stream"},
+ )
+
+ monkeypatch.setattr(
+ ep_mod,
+ "_http_client",
+ httpx.AsyncClient(transport = httpx.MockTransport(http_handler)),
+ )
+
+ async def run():
+ client = _make_client()
+ async for _ in client.stream_chat_completion(
+ messages = [{"role": "user", "content": "hi"}],
+ model = "claude-opus-4-7",
+ temperature = 0.7,
+ top_p = 0.95,
+ max_tokens = 32,
+ compaction_threshold = 150_000,
+ ):
+ pass
+ await client.close()
+
+ _drive(run())
+
+ out = capsys.readouterr().out
+ summary = next(
+ (line for line in out.splitlines() if "Anthropic stream complete" in line),
+ "",
+ )
+ assert "compaction_input_tokens=None" in summary, summary
+ assert "compaction_output_tokens=None" in summary, summary
+
+
+# ── compaction block round-trip (Codex P1) ──────────────────────────
+
+
+def _async_collect(agen):
+ async def run():
+ out = []
+ async for line in agen:
+ out.append(line)
+ return out
+
+ return _drive(run())
+
+
+def test_compaction_block_emitted_as_tool_event(monkeypatch):
+ # Codex P1: once context_management is enabled and Anthropic runs
+ # compaction during a turn, the response carries a
+ # `{type:"compaction", content:""}` block. The translator
+ # must surface it so the chat-adapter can persist it onto the
+ # assistant message; otherwise the next turn loses the state and
+ # Anthropic re-compacts from scratch.
+
+ def http_handler(request: httpx.Request) -> httpx.Response:
+ # Anthropic ships compaction blocks as a content_block_start
+ # with `type:"compaction"`, then either includes the summary
+ # on that start event AND/OR streams it via text_delta events
+ # on the same block index. Test the streamed-delta path since
+ # it's the harder case.
+ body = (
+ b"event: message_start\n"
+ b'data: {"type":"message_start","message":{"usage":{}}}\n\n'
+ b"event: content_block_start\n"
+ b'data: {"type":"content_block_start","index":0,'
+ b'"content_block":{"type":"compaction","content":""}}\n\n'
+ b"event: content_block_delta\n"
+ b'data: {"type":"content_block_delta","index":0,'
+ b'"delta":{"type":"text_delta","text":"Summary so far: "}}\n\n'
+ b"event: content_block_delta\n"
+ b'data: {"type":"content_block_delta","index":0,'
+ b'"delta":{"type":"text_delta","text":"user asked about caching."}}\n\n'
+ b"event: content_block_stop\n"
+ b'data: {"type":"content_block_stop","index":0}\n\n'
+ b"event: content_block_start\n"
+ b'data: {"type":"content_block_start","index":1,'
+ b'"content_block":{"type":"text","text":""}}\n\n'
+ b"event: content_block_delta\n"
+ b'data: {"type":"content_block_delta","index":1,'
+ b'"delta":{"type":"text_delta","text":"Here is my answer."}}\n\n'
+ b"event: content_block_stop\n"
+ b'data: {"type":"content_block_stop","index":1}\n\n'
+ b"event: message_delta\n"
+ b'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},'
+ b'"usage":{"input_tokens":100,"output_tokens":10}}\n\n'
+ b"event: message_stop\n"
+ b'data: {"type":"message_stop"}\n\n'
+ )
+ return httpx.Response(
+ 200,
+ content = body,
+ headers = {"content-type": "text/event-stream"},
+ )
+
+ monkeypatch.setattr(
+ ep_mod,
+ "_http_client",
+ httpx.AsyncClient(transport = httpx.MockTransport(http_handler)),
+ )
+
+ client = _make_client()
+ lines = _async_collect(
+ client._stream_anthropic(
+ messages = [{"role": "user", "content": "hi"}],
+ model = "claude-opus-4-7",
+ temperature = 0.7,
+ top_p = 0.95,
+ max_tokens = 1024,
+ compaction_threshold = 150_000,
+ )
+ )
+ _drive(client.close())
+
+ # Pull tool_events out of the SSE stream and check for the
+ # compaction_block payload.
+ events = []
+ for line in lines:
+ if not line.startswith("data:"):
+ continue
+ raw = line[len("data:") :].strip()
+ if not raw or raw == "[DONE]":
+ continue
+ try:
+ parsed = json.loads(raw)
+ except json.JSONDecodeError:
+ continue
+ # tool_event payloads ride inside chat.completion.chunk.choices[0].delta.content
+ # as a JSON-encoded string. The simpler path: look for the
+ # marker substring anywhere in the chunk.
+ if "compaction_block" in raw:
+ events.append(raw)
+ assert events, f"no compaction_block tool event found in {lines}"
+ # The summary text must come through intact.
+ payload = events[0]
+ assert "Summary so far: user asked about caching." in payload, payload
+
+ # The user-visible content stream must NOT carry the compaction
+ # summary -- only the assistant prose ("Here is my answer.").
+ content_text = ""
+ for line in lines:
+ if not line.startswith("data:"):
+ continue
+ raw = line[len("data:") :].strip()
+ if not raw or raw == "[DONE]":
+ continue
+ try:
+ parsed = json.loads(raw)
+ except json.JSONDecodeError:
+ continue
+ if parsed.get("object") != "chat.completion.chunk":
+ continue
+ for choice in parsed.get("choices") or []:
+ delta = choice.get("delta") or {}
+ chunk = delta.get("content")
+ if isinstance(chunk, str):
+ content_text += chunk
+ assert "Summary so far" not in content_text, content_text
+ assert "Here is my answer." in content_text, content_text
+
+
+def test_compaction_block_round_trips_through_outbound_messages(monkeypatch):
+ # Once the prior turn persisted a compaction block onto the
+ # assistant message, the next turn's outbound body must forward
+ # the {type:"compaction", content:"..."} block to Anthropic
+ # verbatim so the API recognises the existing state.
+ captured: dict = {}
+
+ def http_handler(request: httpx.Request) -> httpx.Response:
+ captured["body"] = json.loads(request.content.decode("utf-8"))
+ return httpx.Response(
+ 200,
+ content = b'event: message_stop\ndata: {"type": "message_stop"}\n\n',
+ headers = {"content-type": "text/event-stream"},
+ )
+
+ monkeypatch.setattr(
+ ep_mod,
+ "_http_client",
+ httpx.AsyncClient(transport = httpx.MockTransport(http_handler)),
+ )
+
+ client = _make_client()
+
+ async def run():
+ async for _ in client.stream_chat_completion(
+ messages = [
+ {"role": "user", "content": "turn 1 question"},
+ {
+ "role": "assistant",
+ "content": [
+ {
+ "type": "compaction",
+ "content": "PRIOR SUMMARY: user asked about caching.",
+ },
+ {"type": "text", "text": "Sure, here's an answer."},
+ ],
+ },
+ {"role": "user", "content": "turn 2 follow-up"},
+ ],
+ model = "claude-opus-4-7",
+ temperature = 0.7,
+ top_p = 0.95,
+ max_tokens = 32,
+ compaction_threshold = 150_000,
+ ):
+ pass
+
+ _drive(run())
+ _drive(client.close())
+
+ msgs = captured["body"]["messages"]
+ # The assistant turn must include the compaction block on the wire.
+ assistant = next((m for m in msgs if m["role"] == "assistant"), None)
+ assert assistant is not None, msgs
+ parts = assistant["content"]
+ types = [p.get("type") for p in parts if isinstance(p, dict)]
+ assert "compaction" in types, parts
+ compaction_part = next(p for p in parts if p.get("type") == "compaction")
+ assert compaction_part["content"] == "PRIOR SUMMARY: user asked about caching."
+
+
+def test_compaction_content_part_accepted_by_chat_message_schema():
+ # Without this Pydantic Tag the discriminated Union would 422 the
+ # request at parse time and the round-trip would never reach the
+ # translator.
+ from models.inference import ChatMessage
+
+ msg = ChatMessage.model_validate(
+ {
+ "role": "assistant",
+ "content": [
+ {"type": "compaction", "content": "summary text"},
+ {"type": "text", "text": "answer prose"},
+ ],
+ }
+ )
+ assert isinstance(msg.content, list)
+ assert msg.content[0].type == "compaction"
+ assert msg.content[0].content == "summary text"
+ assert msg.content[1].type == "text"
+
+
+def test_build_external_messages_passes_compaction_for_anthropic_only():
+ # Compaction is an Anthropic-only synthetic content part. The
+ # builder MUST gate it on provider_type=="anthropic"; every other
+ # provider would 400 on the unknown content type via generic
+ # /chat/completions passthrough (Codex P1 follow-up).
+ from models.inference import ChatMessage
+ from routes.inference import _build_external_messages
+
+ msgs = [
+ ChatMessage.model_validate(
+ {
+ "role": "assistant",
+ "content": [
+ {"type": "compaction", "content": "prior summary"},
+ {"type": "text", "text": "answer"},
+ ],
+ }
+ )
+ ]
+ out = _build_external_messages(
+ msgs, supports_vision = True, provider_type = "anthropic"
+ )
+ assert len(out) == 1
+ parts = out[0]["content"]
+ assert parts[0] == {"type": "compaction", "content": "prior summary"}
+ assert parts[1] == {"type": "text", "text": "answer"}
+
+
+def test_build_external_messages_strips_compaction_for_non_anthropic_providers():
+ # Provider switch (or reused history) hands compaction blocks to a
+ # non-Anthropic provider. Those land on generic /chat/completions
+ # passthrough where the unknown content type fails the upstream
+ # validator. Builder must strip the part for every non-anthropic
+ # provider, including OpenAI/DeepSeek/Mistral/Gemini/Kimi/OpenRouter.
+ from models.inference import ChatMessage
+ from routes.inference import _build_external_messages
+
+ msgs = [
+ ChatMessage.model_validate(
+ {
+ "role": "assistant",
+ "content": [
+ {"type": "compaction", "content": "prior summary"},
+ {"type": "text", "text": "answer"},
+ ],
+ }
+ )
+ ]
+ for provider in ("openai", "deepseek", "mistral", "gemini", "kimi", "openrouter"):
+ out = _build_external_messages(
+ msgs, supports_vision = True, provider_type = provider
+ )
+ assert len(out) == 1, (provider, out)
+ parts = out[0]["content"]
+ types = [p.get("type") for p in parts if isinstance(p, dict)]
+ assert "compaction" not in types, (provider, parts)
+ # Text part survives.
+ assert {"type": "text", "text": "answer"} in parts, (provider, parts)
+
+
+def test_build_external_messages_strips_compaction_when_provider_type_unknown():
+ # Defensive: if provider_type is None (legacy path) the part must
+ # also be stripped -- forwarding to an unknown destination is
+ # never safe.
+ from models.inference import ChatMessage
+ from routes.inference import _build_external_messages
+
+ msgs = [
+ ChatMessage.model_validate(
+ {
+ "role": "assistant",
+ "content": [
+ {"type": "compaction", "content": "prior summary"},
+ {"type": "text", "text": "answer"},
+ ],
+ }
+ )
+ ]
+ out = _build_external_messages(msgs, supports_vision = True)
+ parts = out[0]["content"]
+ types = [p.get("type") for p in parts if isinstance(p, dict)]
+ assert "compaction" not in types, parts
+
+
+def test_build_external_messages_non_vision_anthropic_keeps_compaction():
+ # Defensive: even though compaction-capable Anthropic models all
+ # currently report supports_vision=True, gate the non-vision branch
+ # by provider_type too so future config changes don't drop it.
+ from models.inference import ChatMessage
+ from routes.inference import _build_external_messages
+
+ msgs = [
+ ChatMessage.model_validate(
+ {
+ "role": "assistant",
+ "content": [
+ {"type": "compaction", "content": "prior summary"},
+ {"type": "text", "text": "answer"},
+ ],
+ }
+ )
+ ]
+ out = _build_external_messages(
+ msgs, supports_vision = False, provider_type = "anthropic"
+ )
+ parts = out[0]["content"]
+ assert {"type": "compaction", "content": "prior summary"} in parts
+ # Non-anthropic + non-vision -> compaction stripped, text collapsed
+ # back to a string.
+ out2 = _build_external_messages(
+ msgs, supports_vision = False, provider_type = "deepseek"
+ )
+ assert out2[0]["content"] == "answer", out2
diff --git a/studio/backend/tests/test_anthropic_web_fetch.py b/studio/backend/tests/test_anthropic_web_fetch.py
index 115bfcb500..cdb5f6254c 100644
--- a/studio/backend/tests/test_anthropic_web_fetch.py
+++ b/studio/backend/tests/test_anthropic_web_fetch.py
@@ -157,9 +157,14 @@ def test_web_fetch_combined_with_web_search_and_code_execution(monkeypatch):
tools = captured["body"].get("tools") or []
tool_types = [t.get("type") for t in tools]
- assert "web_search_20250305" in tool_types
- assert "web_fetch_20250910" in tool_types
- assert "code_execution_20250825" in tool_types
+ # After PR 5679's per-model tool version dispatch landed,
+ # claude-opus-4-7 routes web_search to the _20260209 variant and
+ # code_execution to _20260120. web_fetch still hardcodes
+ # _20250910 today; see follow-up to thread it through
+ # _anthropic_web_fetch_version.
+ assert "web_search_20260209" in tool_types, tool_types
+ assert "web_fetch_20250910" in tool_types, tool_types
+ assert "code_execution_20260120" in tool_types, tool_types
# Code-execution still adds its beta flag; web_fetch must not
# have accidentally stripped it.
assert "code-execution-2025-08-25" in captured["headers"].get("anthropic-beta", "")