9a737facaf* 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>
648 lines
23 KiB
Python
648 lines
23 KiB
Python
# 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
|