From 9a737facaf07a656561ea40258eb17a8646cb11c Mon Sep 17 00:00:00 2001 From: Daniel Han Date: Fri, 22 May 2026 06:19:09 -0700 Subject: [PATCH] Studio: wire Anthropic server-side context compaction (#5686) * 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:""}` 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> --- .../core/inference/external_provider.py | 222 +++++- studio/backend/models/inference.py | 40 ++ studio/backend/routes/inference.py | 44 +- .../tests/test_anthropic_compaction.py | 648 ++++++++++++++++++ .../backend/tests/test_anthropic_web_fetch.py | 11 +- 5 files changed, 927 insertions(+), 38 deletions(-) create mode 100644 studio/backend/tests/test_anthropic_compaction.py 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", "")