ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Studio: wire Anthropic server-side context compaction (#5686)

Browse source 
* Studio: wire Anthropic server-side context compaction

Anthropic ships server-side context compaction as a beta
(`compact-2026-01-12`). When the rendered prompt crosses the
configured input-token threshold, Anthropic runs an extra LLM pass
that summarises older turns and the request continues against the
compacted prefix. The response carries the original top-level fields
plus a new `context_management` block (with `applied_edits`) and
`usage.iterations[]` accounting per pass.

Per the docs the feature is currently supported on Opus 4.6, Opus 4.7,
Sonnet 4.6, and Mythos preview. The minimum threshold is 50k tokens;
under-50k requests 400.

Changes:

- Add prefix gate + helper `_anthropic_supports_compaction` plus
  constants `_ANTHROPIC_COMPACTION_PREFIXES`, `_ANTHROPIC_COMPACTION_BETA`,
  `_ANTHROPIC_COMPACTION_TYPE`, `_ANTHROPIC_COMPACTION_MIN`.
- Add `compaction_threshold: Optional[int]` to ChatCompletionRequest
  (50k ge bound, 2M le bound). Thread through `routes/inference.py`
  -> `stream_chat_completion` -> `_stream_anthropic`.
- In `_stream_anthropic`, when threshold is set AND the model
  accepts compaction, attach `context_management.edits[{type:
  "compact_20260112", trigger:{type:"input_tokens", value:N}}]` to
  the outbound body. Sub-50k values are clamped up to 50k to keep
  the request well-formed.
- Refactor the anthropic-beta header builder to merge any combination
  of `code-execution-2025-08-25` + `compact-2026-01-12` flags into
  one header value. Unrelated betas added at the registry level still
  pass through.
- Add `test_anthropic_compaction.py` with 16 cases: gate matrix
  (every doc-listed model), correct body shape, threshold clamping,
  beta header merge with code execution, silent no-op on unsupported
  models, omitted-threshold pass-through.

Live verified end-to-end against the real Anthropic API:
`compact_20260112` accepted on Opus 4.7, response carries
`context_management.applied_edits` + `usage.iterations[]` as
documented. (The first WebFetch-summarised version of these docs
suggested `compact_20260120`; the actual API only accepts
`compact_20260112`, matching the beta-header date. Worth pinning
behind a test so a future doc update can't drift back.)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Address review: drop ge=50_000 clamp + parse usage.iterations[]

Two reviewer follow-ups on the compaction PR:

1. Pydantic ge=50_000 on compaction_threshold was dead code.
   FastAPI rejected sub-50k threshold values with a 422 before the
   `max(int(...), _ANTHROPIC_COMPACTION_MIN)` clamp in
   _stream_anthropic could ever fire. Relaxed the floor to ge=1 so
   the in-helper clamp actually does its job; the schema comment
   now explains why this is intentional. Added a regression test
   that posts a value of 1 and 49_999 through the real request
   schema.

2. Anthropic publishes per-iteration token counts in
   `usage.iterations[]` whenever a fresh compaction has run, and
   the top-level input_tokens / output_tokens cover only the
   `message` iteration -- billing must add the compaction
   iterations on top. Aggregate compaction iteration tokens into
   `last_usage["compaction_input_tokens" / "compaction_output_tokens"]`
   so the cost surface (PR 5690) can read them without re-walking
   the array, and surface both figures in the closing stream
   summary log. Added two tests: one that pins the aggregation on a
   compacted turn and one that pins `None` when no fresh
   iterations land (so re-applied compaction blocks don't double-bill).

Sourcing: https://platform.claude.com/docs/en/build-with-claude/compaction

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Address review: round-trip Anthropic compaction blocks across turns

Codex P1: once context_management is enabled and Anthropic runs
server-side compaction mid-stream, the response carries a
`{type:"compaction", content:"<summary>"}` content block on the
assistant message. The translator only handled text_delta and
input_json_delta on content_block_delta, so the compaction block
was silently dropped. Worse, the request schema's ContentPart
discriminated Union didn't accept `type:"compaction"`, and
_build_external_messages didn't pass it through, so even a
hand-crafted assistant message carrying the block would 422 at
parse time. Net result: Anthropic re-compacted from scratch on
every subsequent turn, wasting input tokens and reasoning budget.

End-to-end backend wiring of the round-trip:

1. SSE translator. _stream_anthropic now tracks a `current_compaction`
   state slot. content_block_start with type=="compaction" seeds it
   (Anthropic may include the summary on the start event AND/OR
   stream it via text_delta events on the same block index --
   handle both). text_delta inside a compaction block routes into
   the compaction buffer instead of the user-visible content
   stream, since the summary is opaque internal state, not
   assistant prose. content_block_stop emits a `compaction_block`
   tool_event carrying the full summary so the chat-adapter can
   persist it. compaction_blocks_seen is surfaced in the closing
   summary log.

2. Pydantic schema. Added CompactionContentPart with Tag("compaction")
   on the ContentPart Union so requests carrying the block parse
   cleanly. Required `content` field with a docstring pointing at
   the Anthropic docs.

3. Message builder. _build_external_messages forwards compaction
   parts on both vision and non-vision paths; the per-provider
   stream helper decides whether to forward to the wire (Anthropic
   does; other providers ignore the part). When a non-vision route
   ends up with a single text part, collapse back to a string
   so providers that don't accept content arrays still get the
   expected shape.

4. _stream_anthropic outbound translator. {type:"compaction"} parts
   on an assistant message land on the wire verbatim. Empty/missing
   `content` is skipped so a malformed stored block can't 400
   Anthropic.

Tests added (5): stream emits compaction_block tool event with the
summary intact; user-visible content stream does NOT carry the
summary text; outbound body forwards compaction parts verbatim on
the next turn; Pydantic schema accepts the part; builder passes
it through on both vision and non-vision provider routes.

Frontend follow-up: the chat-adapter needs to persist the
compaction_block tool_event onto the stored assistant message so
turn N+1 includes it in payload.messages. Pinned in the PR
description.

Sourcing: https://platform.claude.com/docs/en/build-with-claude/compaction

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Address review: gate compaction-part passthrough to Anthropic only

Codex P1: my previous round-trip change preserved {type:"compaction"}
parts on every provider route in _build_external_messages. That
meant a chat history with prior compaction state silently leaked
the Anthropic-specific block to OpenAI/DeepSeek/Mistral/Gemini/
Kimi/OpenRouter on a provider switch, where generic
/chat/completions passthrough hands the unknown content type to
the upstream API and 400s the whole turn.

Added a `provider_type` kwarg to _build_external_messages and
gated the compaction forwarder on `provider_type == "anthropic"`.
Every other value (including the legacy None for callers that
don't pass it yet) strips the part. The Anthropic stream helper
still maps it to a native `compaction` block on the wire.

Threaded provider_type through from _proxy_to_external_provider's
call site.

Tests updated: vision + provider="anthropic" still forwards; six
non-anthropic providers strip the part; missing provider_type
strips defensively; non-vision + anthropic still forwards; non-vision
+ non-anthropic collapses back to a text string.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Daniel Han 2026-05-22 06:19:09 -07:00 • committed by GitHub
commit 9a737facaf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 927 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

222
studio/backend/core/inference/external_provider.py
View file

@ -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
# <think> 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("</think>")
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
# <think> 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("</think>")
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()

40
studio/backend/models/inference.py
View file

@ -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": "<summary>"}`` 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 = (

44
studio/backend/routes/inference.py
View file

@ -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:

648
studio/backend/tests/test_anthropic_compaction.py Normal file
View file

@ -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:"<summary>"}` 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

11
studio/backend/tests/test_anthropic_web_fetch.py
View file

@ -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", "")