diff --git a/tests/integration/_support/claude_code.py b/tests/integration/_support/claude_code.py new file mode 100644 index 00000000000..044713e9656 --- /dev/null +++ b/tests/integration/_support/claude_code.py @@ -0,0 +1,796 @@ +"""Shared Claude Code-shaped request builders and upstream stream fixtures for integration contracts.""" + +import json +from collections.abc import Mapping +from itertools import chain +from typing import Final + +from pydantic import JsonValue, TypeAdapter + +JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) +ANTHROPIC_API_KEY: Final = "synthetic-anthropic-key" +FABLE: Final = "claude-fable-5-1" +OPUS: Final = "claude-opus-5-5" +CLI_BETA: Final = ( + "claude-code-20250219,interleaved-thinking-2025-05-14,thinking-token-count-2026-05-13," + "context-management-2025-06-27,prompt-caching-scope-2026-01-05" +) +FRONTIER_CLI_BETA: Final = ( + f"{CLI_BETA},mid-conversation-system-2026-04-07,per-turn-control-2026-07-01," + "mid-conversation-tool-changes-2026-07-01,effort-2025-11-24" +) +CACHE: Final = {"type": "ephemeral"} +CONTEXT_MANAGEMENT: Final = {"edits": [{"type": "clear_thinking_20251015", "keep": "all"}]} +THINKING_BUDGET: Final = {"budget_tokens": 31999, "type": "enabled", "display": "omitted"} +THINKING_ADAPTIVE: Final = {"type": "adaptive", "display": "omitted"} +METADATA_USER_ID: Final = json.dumps( + { + "device_id": "0" * 64, + "account_uuid": "", + "session_id": "00000000-0000-4000-8000-000000000000", + } +) + + +def schema(properties: JsonValue, required: tuple[str, ...]) -> dict[str, JsonValue]: + return { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": properties, + "required": list(required), + "additionalProperties": False, + } + + +def field(description: str, **extra: JsonValue) -> dict[str, JsonValue]: + return {"description": description, **extra} + + +def tools() -> tuple[dict[str, JsonValue], ...]: + _MAX: Final = 9007199254740991 + return ( + { + "name": "Agent", + "description": "Launch a new agent to handle complex, multi-step tasks.", + "input_schema": schema( + { + "description": field("A short (3-5 word) description of the task", type="string"), + "prompt": field("The task for the agent to perform", type="string"), + "subagent_type": field("The type of specialized agent to use for this task", type="string"), + "model": field( + "Optional model override for this agent.", + type="string", + enum=["sonnet", "opus", "haiku", "fable"], + ), + "run_in_background": field( + "Agents run in the background by default; you will be notified when one completes.", + type="boolean", + ), + "isolation": field("Isolation mode.", type="string", enum=["worktree", "remote"]), + }, + ("description", "prompt"), + ), + }, + { + "name": "Bash", + "description": "Executes a given bash command and returns its output.", + "input_schema": schema( + { + "command": field("The command to execute", type="string"), + "timeout": field("Optional timeout in milliseconds (max 600000)", type="number"), + "description": field( + "Clear, concise description of what this command does in active voice.", + type="string", + ), + "run_in_background": field("Set to true to run this command in the background.", type="boolean"), + "dangerouslyDisableSandbox": field( + "Set this to true to dangerously override sandbox mode and run commands without sandboxing.", + type="boolean", + ), + }, + ("command",), + ), + }, + { + "name": "CronCreate", + "description": "Schedule a prompt to be enqueued at a future time.", + "input_schema": schema( + { + "cron": field( + 'Standard 5-field cron expression in local time: "M H DoM Mon DoW" (e.g.', + type="string", + ), + "prompt": field("The prompt to enqueue at each fire time.", type="string"), + "recurring": field( + "true (default) = fire on every cron match until deleted or auto-expired after 7 days.", + type="boolean", + ), + "durable": field( + "true = persist to .claude/scheduled_tasks.json and survive restarts.", + type="boolean", + ), + }, + ("cron", "prompt"), + ), + }, + { + "name": "CronDelete", + "description": "Cancel a cron job previously scheduled with CronCreate.", + "input_schema": schema( + { + "id": field("Job ID returned by CronCreate.", type="string"), + }, + ("id",), + ), + }, + { + "name": "CronList", + "description": "List all cron jobs scheduled via CronCreate, both durable (.claude/scheduled_tasks.json) and session-only.", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "Edit", + "description": "Performs exact string replacements in files.", + "input_schema": schema( + { + "file_path": field("The absolute path to the file to modify", type="string"), + "old_string": field("The text to replace", type="string"), + "new_string": field( + "The text to replace it with (must be different from old_string)", type="string" + ), + "replace_all": field( + "Replace all occurrences of old_string (default false)", + default=False, + type="boolean", + ), + }, + ("file_path", "old_string", "new_string"), + ), + }, + { + "name": "EnterWorktree", + "description": "Use this tool ONLY when explicitly instructed to work in a worktree — either by the user directly, or by project instruc", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "name": field("Optional name for a new worktree.", type="string"), + "path": field( + "Path to an existing worktree to switch into instead of creating a new one.", + type="string", + ), + }, + "additionalProperties": False, + }, + }, + { + "name": "ExitWorktree", + "description": "Exit a worktree session created by EnterWorktree and return the session to the original working directory.", + "input_schema": schema( + { + "action": field( + '"keep" leaves the worktree and branch on disk; "remove" deletes both.', + type="string", + enum=["keep", "remove"], + ), + "discard_changes": field( + 'Required true when action is "remove" and the worktree has uncommitted files or unmerged commits.', + type="boolean", + ), + }, + ("action",), + ), + }, + { + "name": "ListAgents", + "description": "Lists agents you can SendMessage to — in-process subagents you spawned, the teammates on your team, other local Claude s", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "channel": field("Not available in this build; leave unset.", type="string", maxLength=256), + "q": field("Not available in this build; leave unset.", type="string", maxLength=256), + }, + "additionalProperties": False, + }, + }, + { + "name": "NotebookEdit", + "description": "Replaces, inserts, or deletes a single cell in a Jupyter notebook (.ipynb file).", + "input_schema": schema( + { + "notebook_path": field( + "The absolute path to the Jupyter notebook file to edit (must be absolute, not relative)", + type="string", + ), + "cell_id": field("The ID of the cell to edit.", type="string"), + "new_source": field("The new source for the cell", type="string"), + "cell_type": field( + "The type of the cell (code or markdown).", + type="string", + enum=["code", "markdown"], + ), + "edit_mode": field( + "The type of edit to make (replace, insert, delete).", + type="string", + enum=["replace", "insert", "delete"], + ), + }, + ("notebook_path", "new_source"), + ), + }, + { + "name": "Read", + "description": "Reads a file from the local filesystem.", + "input_schema": schema( + { + "file_path": field("The absolute path to the file to read", type="string"), + "offset": field("The line number to start reading from.", type="integer", minimum=0, maximum=_MAX), + "limit": field( + "The number of lines to read.", + type="integer", + exclusiveMinimum=0, + maximum=_MAX, + ), + "pages": field('Page range for PDF files (e.g., "1-5", "3", "10-20").', type="string"), + }, + ("file_path",), + ), + }, + { + "name": "ReportFindings", + "description": "Report code-review findings as a typed list so the host UI can render them.", + "input_schema": schema( + { + "level": field( + "Effort level the review ran at", + type="string", + enum=["low", "medium", "high", "xhigh", "max"], + ), + "findings": field( + "Verified findings, most-severe first; empty if none survived", + maxItems=32, + type="array", + items={ + "type": "object", + "properties": { + "file": field("Repo-relative path of the file the finding is in", type="string"), + "line": field( + "1-indexed line the finding anchors to", + type="integer", + minimum=-_MAX, + maximum=_MAX, + ), + "summary": field("One-sentence statement of the defect", type="string"), + "short_summary": field( + "Compressed label for compact UI (≤60 chars): the claim alone, no rationale or consequence clause", + type="string", + maxLength=60, + ), + "failure_scenario": field("Concrete inputs/state → wrong output/crash", type="string"), + "category": field( + "Short kebab-case slug of the finding type, e.g.", + type="string", + maxLength=40, + ), + "verdict": field( + "Set when a verify pass ran; absent on inline-only reviews", + type="string", + enum=["CONFIRMED", "PLAUSIBLE"], + ), + "outcome": field( + "Set ONLY when re-reporting after applying fixes: what happened to this finding", + type="string", + enum=["fixed", "skipped", "no_change_needed"], + ), + }, + "required": ["file", "summary", "failure_scenario"], + "additionalProperties": False, + }, + ), + }, + ("findings",), + ), + }, + { + "name": "ScheduleWakeup", + "description": "Schedule when to resume work in /loop dynamic mode — the user invoked /loop without an interval, asking you to self-pace", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "delaySeconds": field("Seconds from now to wake up.", type="number"), + "reason": field("One short sentence explaining the chosen delay.", type="string"), + "prompt": field("The /loop input to fire on wake-up.", type="string"), + "stop": field( + "Set to true to end the dynamic loop immediately instead of scheduling another wakeup.", + type="boolean", + ), + "noop": field( + "true = nothing changed (you checked and there is nothing to report).", + type="boolean", + ), + }, + "additionalProperties": False, + }, + }, + { + "name": "SendMessage", + "description": "# SendMessage\n\nSend a message to another agent.", + "input_schema": schema( + { + "to": field( + 'Recipient: a name from ListAgents (append its " [ref]" only when a listing or an error shows one), a teammate name, "mai', + type="string", + allOf=[{"pattern": "^[^\\n\\r]*$"}, {"pattern": "^[\\s\\S]{0,300}$"}], + ), + "summary": field( + "A 5-10 word label for your own transcript row (not transmitted — the recipient previews the first line of `message`).", + type="string", + maxLength=200, + ), + "message": field("Plain text message content.", default="", type="string"), + "notify_when_idle": field( + "Ask a session ON THIS MACHINE to send you ONE notice when it next goes idle (finishes its turn with nothing queued) or e", + type="boolean", + ), + }, + ("to", "message"), + ), + }, + { + "name": "Skill", + "description": "Invoke a skill.", + "input_schema": schema( + { + "skill": field("The name of a skill from the available-skills list.", type="string"), + "args": field("Optional arguments for the skill", type="string"), + }, + ("skill",), + ), + }, + { + "name": "TaskCreate", + "description": "Use this tool to create a structured task list for your current coding session.", + "input_schema": schema( + { + "subject": field("A brief title for the task", type="string"), + "description": field("What needs to be done", type="string"), + "activeForm": field( + 'Present continuous form shown in spinner when in_progress (e.g., "Running tests")', + type="string", + ), + "metadata": field( + "Arbitrary metadata to attach to the task", + type="object", + propertyNames={"type": "string"}, + additionalProperties={}, + ), + }, + ("subject", "description"), + ), + }, + { + "name": "TaskGet", + "description": "Use this tool to retrieve a task by its ID from the task list.", + "input_schema": schema( + { + "taskId": field("The ID of the task to retrieve", type="string"), + }, + ("taskId",), + ), + }, + { + "name": "TaskList", + "description": "Use this tool to list all tasks in the task list.", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "TaskStop", + "description": "- Stops a running background task by its ID\n- Takes a task_id parameter identifying the task to stop\n- To stop an agent-", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "task_id": field("The ID of the background task to stop.", type="string"), + "shell_id": field("Deprecated: use task_id instead", type="string"), + }, + "additionalProperties": False, + }, + }, + { + "name": "TaskUpdate", + "description": "Use this tool to update a task in the task list.", + "input_schema": schema( + { + "taskId": field("The ID of the task to update", type="string"), + "subject": field("New subject for the task", type="string"), + "description": field("New description for the task", type="string"), + "activeForm": field( + 'Present continuous form shown in spinner when in_progress (e.g., "Running tests")', + type="string", + ), + "status": field( + "New status for the task", + anyOf=[ + {"type": "string", "enum": ["pending", "in_progress", "completed"]}, + {"type": "string", "const": "deleted"}, + ], + ), + "addBlocks": field("Task IDs that this task blocks", type="array", items={"type": "string"}), + "addBlockedBy": field("Task IDs that block this task", type="array", items={"type": "string"}), + "owner": field("New owner for the task", type="string"), + "metadata": field( + "Metadata keys to merge into the task.", + type="object", + propertyNames={"type": "string"}, + additionalProperties={}, + ), + }, + ("taskId",), + ), + }, + { + "name": "WebFetch", + "description": "IMPORTANT: WebFetch WILL FAIL for authenticated or private URLs.", + "input_schema": schema( + { + "url": field("The URL to fetch content from", type="string", format="uri"), + "prompt": field("The prompt to run on the fetched content", type="string"), + }, + ("url", "prompt"), + ), + }, + { + "name": "WebSearch", + "description": "- Allows Claude to search the web and use the results to inform responses\n- Provides up-to-date information for current ", + "input_schema": schema( + { + "query": field("The search query to use", type="string", minLength=2), + "allowed_domains": field( + "Only include search results from these domains", type="array", items={"type": "string"} + ), + "blocked_domains": field( + "Never include search results from these domains", type="array", items={"type": "string"} + ), + }, + ("query",), + ), + }, + { + "name": "Workflow", + "description": "Execute a workflow script that orchestrates multiple subagents deterministically.", + "input_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "script": field("Self-contained workflow script.", type="string", maxLength=524288), + "name": field( + "Name of a predefined workflow (built-in or from .claude/workflows/).", type="string" + ), + "description": field( + "Ignored — set the workflow description in the script's `meta` block.", type="string" + ), + "title": field("Ignored — set the workflow title in the script's `meta` block.", type="string"), + "args": field("Optional input value exposed to the script as the global `args`, verbatim."), + "scriptPath": field("Path to a workflow script file on disk.", type="string"), + "resumeFromRunId": field( + "Run ID of a prior Workflow invocation to resume from.", + type="string", + pattern="^wf_[a-z0-9-]{6,}$", + ), + }, + "additionalProperties": False, + }, + }, + { + "name": "Write", + "description": "Writes a file to the local filesystem.", + "input_schema": schema( + { + "file_path": field( + "The absolute path to the file to write (must be absolute, not relative)", type="string" + ), + "content": field("The content to write to the file", type="string"), + }, + ("file_path", "content"), + ), + }, + ) + + +def system_blocks() -> tuple[dict[str, JsonValue], ...]: + return ( + {"type": "text", "text": "x-anthropic-billing-header: cc_version=2.1.283.00; cc_entrypoint=sdk-cli;"}, + {"type": "text", "text": "Synthetic agent identity system prompt.", "cache_control": CACHE}, + {"type": "text", "text": "Synthetic interactive agent instructions.", "cache_control": CACHE}, + ) + + +def claude_code_request(cache_bust: str) -> dict[str, JsonValue]: + reminders: Final = ( + f"\n{cache_bust}\n", + "\nSynthetic model identity reminder.\n", + "\nSynthetic agent types reminder.\n", + "\nSynthetic skills reminder.\n", + "\n15000000 tokens left\n", + "\nSynthetic date reminder.\n", + "\nSynthetic attribution reminder.\n", + ) + return { + "model": "", + "system": list(system_blocks()), + "messages": [ + { + "role": "user", + "content": [ + *[{"type": "text", "text": reminder} for reminder in reminders], + {"type": "text", "text": "Reply with exactly the word PONG", "cache_control": CACHE}, + ], + } + ], + "tools": list(tools()), + "metadata": {"user_id": METADATA_USER_ID}, + "max_tokens": 32000, + "thinking": dict(THINKING_BUDGET), + "context_management": dict(CONTEXT_MANAGEMENT), + "stream": True, + } + + +def frontier_request( + cache_bust: str, + effort: str, + max_tokens: int, + prompt_text: str = "Reply with exactly the word PONG", + stream: bool = True, +) -> dict[str, JsonValue]: + return { + "model": "", + "system": list(system_blocks()), + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": f"\n{cache_bust}\n"}, + {"type": "text", "text": prompt_text}, + ], + }, + { + "role": "system", + "content": [ + { + "type": "text", + "text": "# Environment\nSynthetic environment block.", + "cache_control": CACHE, + } + ], + }, + ], + "tools": list(tools()), + "metadata": {"user_id": METADATA_USER_ID}, + "max_tokens": max_tokens, + "thinking": dict(THINKING_ADAPTIVE), + "context_management": dict(CONTEXT_MANAGEMENT), + "output_config": {"effort": effort}, + "stream": stream, + } + + +def tool_loop_turn2( + base: dict[str, JsonValue], + assistant_content: tuple[dict[str, JsonValue], ...], + tool_results: tuple[tuple[str, JsonValue], ...], +) -> dict[str, JsonValue]: + return { + **base, + "messages": [ + *base["messages"], + {"role": "assistant", "content": list(assistant_content)}, + { + "role": "user", + "content": [ + {"tool_use_id": tool_use_id, "type": "tool_result", "content": content} + for tool_use_id, content in tool_results + ], + }, + { + "role": "system", + "content": [ + { + "type": "text", + "text": "14999970 tokens left", + "cache_control": CACHE, + }, + { + "type": "text", + "text": "First privately list what you need next; then request every item that doesn't depend on another's result in this one response.", + }, + ], + }, + ], + } + + +def cli_headers(key: str, beta: str = CLI_BETA) -> dict[str, str]: + return { + "accept": "application/json", + "content-type": "application/json", + "user-agent": "claude-cli/2.1.283 (external, sdk-cli)", + "x-claude-code-session-id": "00000000-0000-4000-8000-000000000000", + "x-stainless-arch": "x64", + "x-stainless-lang": "js", + "x-stainless-os": "Linux", + "x-stainless-package-version": "0.112.1", + "x-stainless-retry-count": "0", + "x-stainless-runtime": "node", + "x-stainless-runtime-version": "v26.3.0", + "x-stainless-timeout": "600", + "anthropic-beta": beta, + "anthropic-dangerous-direct-browser-access": "true", + "anthropic-version": "2023-06-01", + "x-app": "cli", + "x-api-key": key, + } + + +def sse_frame(event: str, data: JsonValue) -> bytes: + return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode() + + +def sse_events(text: str) -> tuple[tuple[str, dict[str, object]], ...]: + frames: Final = tuple(frame for frame in text.split("\n\n") if frame.strip()) + return tuple( + ( + event, + json.loads(next(line.removeprefix("data: ") for line in frame.splitlines() if line.startswith("data: "))), + ) + for frame in frames + if (event := next(line.removeprefix("event: ") for line in frame.splitlines() if line.startswith("event: "))) + != "ping" + ) + + +def _start_usage(usage: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + return {key: value for key, value in usage.items() if key != "output_tokens"} + + +def text_stream(identity: str, model: str, text: str, usage: dict[str, int]) -> tuple[bytes, ...]: + return ( + sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": model, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": _start_usage(usage), + }, + }, + ), + sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}}, + ), + sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": usage["output_tokens"]}, + }, + ), + sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def _tool_use_frames(index: int, tool_id: str, name: str, tool_input: JsonValue) -> tuple[bytes, ...]: + arguments: Final = json.dumps(tool_input) + return ( + sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": index, + "content_block": {"type": "tool_use", "id": tool_id, "name": name, "input": {}}, + }, + ), + sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": index, + "delta": {"type": "input_json_delta", "partial_json": arguments[: len(arguments) // 2]}, + }, + ), + sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": index, + "delta": {"type": "input_json_delta", "partial_json": arguments[len(arguments) // 2 :]}, + }, + ), + sse_frame("content_block_stop", {"type": "content_block_stop", "index": index}), + ) + + +def tool_use_stream( + identity: str, + model: str, + thinking: str, + signature: str, + tool_calls: tuple[tuple[str, str, JsonValue], ...], + usage: dict[str, int], +) -> tuple[bytes, ...]: + head: Final = ( + sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": model, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": _start_usage(usage), + }, + }, + ), + sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, + ), + sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": thinking}}, + ), + sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": signature}}, + ), + sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + ) + tail: Final = ( + sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": usage["output_tokens"]}, + }, + ), + sse_frame("message_stop", {"type": "message_stop"}), + ) + frames: Final = ( + *head, + *chain.from_iterable( + _tool_use_frames(index, tool_id, name, tool_input) + for index, (tool_id, name, tool_input) in enumerate(tool_calls, start=1) + ), + *tail, + ) + return frames diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_adaptive_thinking_effort_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_adaptive_thinking_effort_wire.py new file mode 100644 index 00000000000..9f929d69a05 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_adaptive_thinking_effort_wire.py @@ -0,0 +1,111 @@ +import uuid +from typing import Final + +import pytest +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + + +def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + key: {"expected": expected.get(key), "upstream": body.get(key)} + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + + +def test_adaptive_thinking_and_effort_reach_anthropic_intact(gateway: Gateway) -> None: + identity: Final = f"msg_fable_{uuid.uuid4().hex}" + request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "high", 64000) + cli_beta: Final = frozenset(cc.FRONTIER_CLI_BETA.split(",")) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY + assert request.headers["anthropic-version"] == "2023-06-01" + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert cli_beta <= frozenset(upstream_beta.split(",")), upstream_beta + assert upstream_beta.split(",").count("effort-2025-11-24") == 1, upstream_beta + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": cc.FABLE} + assert body == expected, _diff(expected, body) + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity, cc.FABLE, "PONG", {"input_tokens": 12, "output_tokens": 4}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + events: Final = cc.sse_events(response.text) + assert [event for event, _ in events][-1] == "message_stop" + assert events[4][1]["delta"]["stop_reason"] == "end_turn" + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["prompt_tokens"] == 12 and rows[0]["completion_tokens"] == 4 + + +def test_xhigh_effort_reaches_anthropic_and_charges_by_usage(gateway: Gateway) -> None: + identity: Final = f"msg_opus_{uuid.uuid4().hex}" + request_body: Final = cc.frontier_request(f"cache-bust-{uuid.uuid4().hex}", "xhigh", 128000) + cli_beta: Final = frozenset(cc.FRONTIER_CLI_BETA.split(",")) + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + assert request.headers["x-api-key"] == cc.ANTHROPIC_API_KEY + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert cli_beta <= frozenset(upstream_beta.split(",")), upstream_beta + assert upstream_beta.split(",").count("effort-2025-11-24") == 1, upstream_beta + body: Final = cc.JSON_OBJECT.validate_json(request.body) + expected: Final = {**request_body, "model": cc.OPUS} + assert body == expected, _diff(expected, body) + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity, cc.OPUS, "PONG", {"input_tokens": 10, "output_tokens": 5}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{cc.OPUS}", + api_base=wire.url, + api_key=cc.ANTHROPIC_API_KEY, + input_cost_per_token=1e-6, + output_cost_per_token=5e-6, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + {**request_body, "model": model}, + params={"beta": "true"}, + headers=cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA), + ) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, prompt_tokens, completion_tokens FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(10 * 1e-6 + 5 * 5e-6) diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_interleaved_thinking_history_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_interleaved_thinking_history_wire.py new file mode 100644 index 00000000000..6990f33e51c --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_interleaved_thinking_history_wire.py @@ -0,0 +1,167 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +_TURN1_TOOL: Final = ("toolu_a", "Read", {"file_path": "/tmp/cc_probe/a.txt"}) +_TURN2_TOOL: Final = ("toolu_b", "Read", {"file_path": "/tmp/cc_probe/b.txt"}) + + +def _diff(expected: dict[str, JsonValue], body: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + key: {"expected": expected.get(key), "upstream": body.get(key)} + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + + +def _turn2(base: dict[str, JsonValue]) -> dict[str, JsonValue]: + return cc.tool_loop_turn2( + base, + ( + {"type": "thinking", "thinking": "plan", "signature": "sig1"}, + {"type": "tool_use", "id": _TURN1_TOOL[0], "name": _TURN1_TOOL[1], "input": _TURN1_TOOL[2]}, + ), + ((_TURN1_TOOL[0], "ALPHA"),), + ) + + +def _turn3(turn2: dict[str, JsonValue]) -> dict[str, JsonValue]: + return cc.tool_loop_turn2( + turn2, + ( + {"type": "thinking", "thinking": "got A", "signature": "sig2"}, + {"type": "text", "text": "got A"}, + {"type": "tool_use", "id": _TURN2_TOOL[0], "name": _TURN2_TOOL[1], "input": _TURN2_TOOL[2]}, + ), + ((_TURN2_TOOL[0], "BRAVO"),), + ) + + +def _interleaved_stream(identity: str) -> tuple[bytes, ...]: + usage: Final = {"input_tokens": 20, "output_tokens": 12} + return ( + cc.sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": cc.FABLE, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": usage["input_tokens"], "output_tokens": 1}, + }, + }, + ), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "got A"}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig2"}}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 1, "content_block": {"type": "text", "text": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "got A"}}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 1}), + cc.sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": 2, + "content_block": {"type": "tool_use", "id": _TURN2_TOOL[0], "name": _TURN2_TOOL[1], "input": {}}, + }, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 2, + "delta": {"type": "input_json_delta", "partial_json": '{"file_path": "/tmp/cc_pr'}, + }, + ), + cc.sse_frame( + "content_block_delta", + { + "type": "content_block_delta", + "index": 2, + "delta": {"type": "input_json_delta", "partial_json": 'obe/b.txt"}'}, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 2}), + cc.sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": usage["output_tokens"]}, + }, + ), + cc.sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def test_interleaved_thinking_text_and_tool_use_history_reaches_anthropic_identical(gateway: Gateway) -> None: + identity: Final = f"msg_il_{uuid.uuid4().hex}" + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/a.txt then /tmp/cc_probe/b.txt one at a time and reply with both words", + ) + turn2: Final = _turn2(turn1) + turn3: Final = _turn3(turn2) + turn2_expected: Final = {**turn2, "model": cc.FABLE} + turn3_expected: Final = {**turn3, "model": cc.FABLE} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + upstream_beta: Final = request.headers.get("anthropic-beta", "") + assert upstream_beta.split(",").count("interleaved-thinking-2025-05-14") == 1, upstream_beta + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body == turn2_expected: + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream("msg_il_turn2", cc.FABLE, "got A", {"input_tokens": 20, "output_tokens": 4}), + ) + assert body == turn3_expected, _diff(turn3_expected, body) + return Reply(content_type="text/event-stream", chunks=_interleaved_stream(identity)) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response2: Final = gateway.request( + "POST", "/v1/messages", {**turn2, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + response3: Final = gateway.request( + "POST", "/v1/messages", {**turn3, "model": model}, params={"beta": "true"}, headers=headers + ) + assert response3.status_code == 200, response3.text + events: Final = cc.sse_events(response3.text) + started: Final = [ + (data["index"], data["content_block"]["type"]) for event, data in events if event == "content_block_start" + ] + assert started == [(0, "thinking"), (1, "text"), (2, "tool_use")], started + assert events[-1][0] == "message_stop" + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (turn2_expected, turn3_expected), bodies diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_model_switch_history_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_model_switch_history_wire.py new file mode 100644 index 00000000000..5753d395bc4 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_model_switch_history_wire.py @@ -0,0 +1,83 @@ +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + + +def test_mid_loop_model_switch_replays_history_byte_identical(gateway: Gateway) -> None: + identity1: Final = f"msg_sw1_{uuid.uuid4().hex}" + identity2: Final = f"msg_sw2_{uuid.uuid4().hex}" + turn1: Final = cc.frontier_request( + f"cache-bust-{uuid.uuid4().hex}", + "high", + 64000, + prompt_text="Read /tmp/cc_probe/hello.txt and reply with its single word", + ) + turn2: Final = cc.tool_loop_turn2( + turn1, + ( + {"type": "thinking", "thinking": "need to read the file", "signature": "sig_anthropic_1"}, + { + "type": "tool_use", + "id": "toolu_read_1", + "name": "Read", + "input": {"file_path": "/tmp/cc_probe/hello.txt"}, + }, + ), + (("toolu_read_1", "1\tPROBE\n2\t"),), + ) + first_expected: Final = {**turn1, "model": cc.FABLE} + second_expected: Final = {**turn2, "model": cc.OPUS} + + def respond(request: Request) -> Reply: + assert request.method == "POST" + assert request.target == "/v1/messages", request.target + body: Final = cc.JSON_OBJECT.validate_json(request.body) + if body == first_expected: + return Reply( + content_type="text/event-stream", + chunks=cc.tool_use_stream( + identity1, + cc.FABLE, + "need to read the file", + "sig_anthropic_1", + (("toolu_read_1", "Read", {"file_path": "/tmp/cc_probe/hello.txt"}),), + {"input_tokens": 20, "output_tokens": 10}, + ), + ) + assert body == second_expected, { + key: (second_expected.get(key), body.get(key)) + for key in second_expected.keys() | body.keys() + if second_expected.get(key) != body.get(key) + } + return Reply( + content_type="text/event-stream", + chunks=cc.text_stream(identity2, cc.OPUS, "PROBE", {"input_tokens": 30, "output_tokens": 3}), + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + fable: Final = scenario.model(model=f"anthropic/{cc.FABLE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + opus: Final = scenario.model(model=f"anthropic/{cc.OPUS}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + headers: Final = cc.cli_headers(gateway.key, cc.FRONTIER_CLI_BETA) + response1: Final = gateway.request( + "POST", "/v1/messages", {**turn1, "model": fable}, params={"beta": "true"}, headers=headers + ) + assert response1.status_code == 200, response1.text + response2: Final = gateway.request( + "POST", "/v1/messages", {**turn2, "model": opus}, params={"beta": "true"}, headers=headers + ) + assert response2.status_code == 200, response2.text + bodies: Final = tuple(cc.JSON_OBJECT.validate_json(request.body) for request in wire.drain()) + assert bodies == (first_expected, second_expected), bodies + rows: Final = eventually( + lambda: read_rows( + 'SELECT model FROM "LiteLLM_SpendLogs" WHERE request_id=%s', + (identity2,), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["model"] == f"anthropic/{cc.OPUS}" diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_request_translation_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_request_translation_wire.py new file mode 100644 index 00000000000..f15cabda823 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_request_translation_wire.py @@ -0,0 +1,366 @@ +import json +import uuid +from collections.abc import Mapping +from typing import Final + +import pytest +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +_HAIKU_4_5: Final = "claude-haiku-4-5" +_OPUS_4_5: Final = "claude-opus-4-5" +_OPUS_4_6: Final = "claude-opus-4-6" +_OPUS_4_7: Final = "claude-opus-4-7" +_FABLE_5_1: Final = "claude-fable-5-1" +_ADAPTIVE: Final = {"type": "adaptive", "display": "omitted"} +_ADAPTIVE_SUMMARIZED: Final = {"type": "adaptive", "display": "summarized"} + + +def _budget(tokens: int) -> dict[str, JsonValue]: + return {"type": "enabled", "budget_tokens": tokens} + + +def _client_body(**reasoning: JsonValue) -> dict[str, JsonValue]: + base: Final = { + key: value + for key, value in cc.claude_code_request(f"cache-bust-{uuid.uuid4().hex}").items() + if key != "thinking" + } + return {**base, "stream": False, **reasoning} + + +def _without(body: Mapping[str, JsonValue], *keys: str) -> dict[str, JsonValue]: + return {key: value for key, value in body.items() if key not in keys} + + +def _diff(expected: Mapping[str, JsonValue], body: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + return { + key: {"expected": expected.get(key), "upstream": body.get(key)} + for key in expected.keys() | body.keys() + if expected.get(key) != body.get(key) + } + + +def _forwarded_body( + gateway: Gateway, upstream_model: str, client_body: Mapping[str, JsonValue] +) -> dict[str, JsonValue]: + def respond(request: Request) -> Reply: + return Reply( + body=json.dumps( + { + "id": f"msg_{uuid.uuid4().hex}", + "type": "message", + "role": "assistant", + "model": upstream_model, + "content": [{"type": "text", "text": "PONG"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 12, "output_tokens": 4}, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{upstream_model}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY + ) + response: Final = gateway.request("POST", "/v1/messages", {**client_body, "model": model}) + assert response.status_code == 200, response.text + received: Final = wire.drain() + assert len(received) == 1, received + return cc.JSON_OBJECT.validate_json(received[0].body) + + +def _assert_forwarded( + gateway: Gateway, + upstream_model: str, + client_body: dict[str, JsonValue], + expected_changes: Mapping[str, JsonValue], + removed: tuple[str, ...], +) -> None: + expected: Final = {**_without(client_body, *removed), **expected_changes, "model": upstream_model} + body: Final = _forwarded_body(gateway, upstream_model, client_body) + assert body == expected, _diff(expected, body) + + +@pytest.mark.parametrize( + ("upstream_model", "effort", "expected_changes", "removed"), + ( + pytest.param( + _OPUS_4_5, + "high", + {}, + ("thinking",), + id="opus-4.5-keeps-supported-effort-drops-adaptive", + ), + pytest.param( + _OPUS_4_5, + "xhigh", + {"thinking": _budget(8192)}, + ("output_config",), + id="opus-4.5-xhigh-falls-back-to-budget", + ), + pytest.param(_HAIKU_4_5, "low", {"thinking": _budget(1024)}, ("output_config",), id="haiku-4.5-low"), + pytest.param(_HAIKU_4_5, "medium", {"thinking": _budget(2048)}, ("output_config",), id="haiku-4.5-medium"), + pytest.param(_HAIKU_4_5, "high", {"thinking": _budget(4096)}, ("output_config",), id="haiku-4.5-high"), + pytest.param(_HAIKU_4_5, "xhigh", {"thinking": _budget(8192)}, ("output_config",), id="haiku-4.5-xhigh"), + pytest.param(_HAIKU_4_5, "max", {"thinking": _budget(16384)}, ("output_config",), id="haiku-4.5-max"), + pytest.param(_OPUS_4_6, "high", {}, (), id="opus-4.6-adaptive-unchanged"), + pytest.param(_OPUS_4_7, "xhigh", {}, (), id="opus-4.7-adaptive-unchanged"), + ), +) +def test_adaptive_thinking_and_effort_are_reshaped_only_for_models_without_adaptive_thinking( + gateway: Gateway, + upstream_model: str, + effort: str, + expected_changes: dict[str, JsonValue], + removed: tuple[str, ...], +) -> None: + client_body: Final = _client_body(thinking=dict(_ADAPTIVE), output_config={"effort": effort}) + _assert_forwarded(gateway, upstream_model, client_body, expected_changes, removed) + + +def test_adaptive_effort_fallback_budget_is_capped_below_max_tokens(gateway: Gateway) -> None: + client_body: Final = _client_body(thinking=dict(_ADAPTIVE), output_config={"effort": "max"}, max_tokens=4000) + _assert_forwarded(gateway, _HAIKU_4_5, client_body, {"thinking": _budget(3999)}, ("output_config",)) + + +def test_adaptive_effort_fallback_drops_thinking_when_max_tokens_cannot_fit_the_minimum_budget( + gateway: Gateway, +) -> None: + client_body: Final = _client_body(thinking=dict(_ADAPTIVE), output_config={"effort": "high"}, max_tokens=1024) + _assert_forwarded(gateway, _HAIKU_4_5, client_body, {}, ("thinking", "output_config")) + + +@pytest.mark.parametrize( + ("budget_tokens", "effort"), + ( + pytest.param(1024, "low", id="below-medium-threshold"), + pytest.param(2048, "medium", id="medium-threshold"), + pytest.param(4096, "high", id="high-threshold"), + pytest.param(8192, "xhigh", id="xhigh-threshold"), + ), +) +def test_legacy_thinking_budget_becomes_adaptive_effort_on_models_that_reject_budgets( + gateway: Gateway, budget_tokens: int, effort: str +) -> None: + client_body: Final = _client_body(thinking=_budget(budget_tokens)) + _assert_forwarded( + gateway, + _OPUS_4_7, + client_body, + {"thinking": {"type": "adaptive"}, "output_config": {"effort": effort}}, + (), + ) + + +def test_legacy_thinking_translation_keeps_the_callers_effort(gateway: Gateway) -> None: + client_body: Final = _client_body(thinking=_budget(8192), output_config={"effort": "medium"}) + _assert_forwarded(gateway, _OPUS_4_7, client_body, {"thinking": {"type": "adaptive"}}, ()) + + +@pytest.mark.parametrize( + ("upstream_model", "removed"), + ( + pytest.param(_FABLE_5_1, ("thinking",), id="always-on-model-drops-disabled"), + pytest.param(_OPUS_4_7, (), id="other-model-keeps-disabled"), + ), +) +def test_disabled_thinking_is_dropped_only_for_always_on_thinking_models( + gateway: Gateway, upstream_model: str, removed: tuple[str, ...] +) -> None: + client_body: Final = _client_body(thinking={"type": "disabled"}) + _assert_forwarded(gateway, upstream_model, client_body, {}, removed) + + +@pytest.mark.parametrize( + ("reasoning_effort", "effort"), + ( + pytest.param("minimal", "low", id="minimal"), + pytest.param("low", "low", id="low"), + pytest.param("medium", "medium", id="medium"), + pytest.param("high", "high", id="high"), + pytest.param("xhigh", "xhigh", id="xhigh"), + pytest.param("max", "max", id="max"), + ), +) +def test_reasoning_effort_becomes_adaptive_thinking_and_effort_on_adaptive_models( + gateway: Gateway, reasoning_effort: str, effort: str +) -> None: + client_body: Final = _client_body(reasoning_effort=reasoning_effort) + _assert_forwarded( + gateway, + _OPUS_4_7, + client_body, + {"thinking": dict(_ADAPTIVE_SUMMARIZED), "output_config": {"effort": effort}}, + ("reasoning_effort",), + ) + + +@pytest.mark.parametrize( + ("reasoning_effort", "budget_tokens"), + ( + pytest.param("minimal", 1024, id="minimal"), + pytest.param("low", 1024, id="low"), + pytest.param("medium", 2048, id="medium"), + pytest.param("high", 4096, id="high"), + pytest.param("xhigh", 8192, id="xhigh"), + pytest.param("max", 16384, id="max"), + ), +) +def test_reasoning_effort_becomes_a_thinking_budget_on_models_without_adaptive_thinking( + gateway: Gateway, reasoning_effort: str, budget_tokens: int +) -> None: + client_body: Final = _client_body(reasoning_effort=reasoning_effort) + _assert_forwarded(gateway, _HAIKU_4_5, client_body, {"thinking": _budget(budget_tokens)}, ("reasoning_effort",)) + + +def test_reasoning_effort_none_clears_thinking_and_effort(gateway: Gateway) -> None: + client_body: Final = _client_body( + reasoning_effort="none", thinking=dict(_ADAPTIVE), output_config={"effort": "high"} + ) + _assert_forwarded(gateway, _OPUS_4_7, client_body, {}, ("reasoning_effort", "thinking", "output_config")) + + +def test_caller_thinking_wins_over_reasoning_effort(gateway: Gateway) -> None: + client_body: Final = _client_body(reasoning_effort="high", thinking=_budget(2000)) + _assert_forwarded(gateway, _HAIKU_4_5, client_body, {}, ("reasoning_effort",)) + + +def test_caller_effort_wins_over_reasoning_effort(gateway: Gateway) -> None: + client_body: Final = _client_body(reasoning_effort="high", output_config={"effort": "low"}) + _assert_forwarded(gateway, _OPUS_4_7, client_body, {"thinking": dict(_ADAPTIVE_SUMMARIZED)}, ("reasoning_effort",)) + + +def test_reasoning_effort_budget_is_capped_below_max_tokens(gateway: Gateway) -> None: + client_body: Final = _client_body(reasoning_effort="max", max_tokens=4000) + _assert_forwarded(gateway, _HAIKU_4_5, client_body, {"thinking": _budget(3999)}, ("reasoning_effort",)) + + +def test_reasoning_effort_is_dropped_when_max_tokens_cannot_fit_the_minimum_budget(gateway: Gateway) -> None: + client_body: Final = _client_body(reasoning_effort="high", max_tokens=1024) + _assert_forwarded(gateway, _HAIKU_4_5, client_body, {}, ("reasoning_effort",)) + + +@pytest.mark.parametrize( + ("upstream_model", "reasoning", "expected_changes", "removed"), + ( + pytest.param( + _HAIKU_4_5, + {"thinking": dict(_ADAPTIVE), "output_config": {"effort": "high"}}, + {"thinking": _budget(4096)}, + ("output_config", "temperature"), + id="haiku-4.5-effort-translated-to-budget", + ), + pytest.param( + _OPUS_4_5, + {"thinking": dict(_ADAPTIVE), "output_config": {"effort": "high"}}, + {}, + ("thinking", "temperature"), + id="opus-4.5-effort-kept", + ), + pytest.param(_HAIKU_4_5, {"thinking": _budget(2048)}, {}, ("temperature",), id="haiku-4.5-legacy-budget"), + ), +) +def test_non_default_temperature_is_dropped_when_a_non_adaptive_model_thinks( + gateway: Gateway, + upstream_model: str, + reasoning: dict[str, JsonValue], + expected_changes: dict[str, JsonValue], + removed: tuple[str, ...], +) -> None: + client_body: Final = _client_body(temperature=0, **reasoning) + _assert_forwarded(gateway, upstream_model, client_body, expected_changes, removed) + + +@pytest.mark.parametrize( + ("upstream_model", "temperature", "reasoning"), + ( + pytest.param(_HAIKU_4_5, 1, {"thinking": _budget(2048)}, id="temperature-1-with-thinking"), + pytest.param(_HAIKU_4_5, 0, {}, id="temperature-0-without-thinking"), + pytest.param( + _OPUS_4_6, + 0, + {"thinking": dict(_ADAPTIVE), "output_config": {"effort": "high"}}, + id="adaptive-model", + ), + ), +) +def test_temperature_is_kept_when_it_does_not_conflict_with_thinking( + gateway: Gateway, upstream_model: str, temperature: int, reasoning: dict[str, JsonValue] +) -> None: + client_body: Final = _client_body(temperature=temperature, **reasoning) + _assert_forwarded(gateway, upstream_model, client_body, {}, ()) + + +_SIGNED_THINKING: Final = {"type": "thinking", "thinking": "check the config first", "signature": "EqQBCkgIBRABGAIiQL"} +_TOOL_CALL: Final = {"type": "tool_use", "id": "toolu_01", "name": "Read", "input": {"file_path": "/repo/config.yaml"}} +_TOOL_RESULT: Final = {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_01", "content": "ok"}]} + + +def _history_body(assistant_content: tuple[dict[str, JsonValue], ...]) -> dict[str, JsonValue]: + base: Final = _client_body(thinking=_budget(2048)) + first_turn: Final = base["messages"] + assert isinstance(first_turn, list) + return {**base, "messages": [*first_turn, {"role": "assistant", "content": list(assistant_content)}, _TOOL_RESULT]} + + +def _with_assistant_content( + body: Mapping[str, JsonValue], assistant_content: tuple[dict[str, JsonValue], ...] +) -> dict[str, JsonValue]: + messages: Final = body["messages"] + assert isinstance(messages, list) + return { + **body, + "messages": [*messages[:-2], {"role": "assistant", "content": list(assistant_content)}, messages[-1]], + } + + +def test_encrypted_reasoning_from_another_provider_is_stripped_and_anthropic_signed_thinking_is_kept( + gateway: Gateway, +) -> None: + client_body: Final = _history_body( + ( + {"type": "thinking", "thinking": "bridge reasoning", "signature": "litellm_encrypted_reasoning:gAAAAB"}, + {"type": "redacted_thinking", "data": "litellm_encrypted_reasoning:gAAAAC"}, + _SIGNED_THINKING, + _TOOL_CALL, + ) + ) + expected: Final = {**_with_assistant_content(client_body, (_SIGNED_THINKING, _TOOL_CALL)), "model": _HAIKU_4_5} + body: Final = _forwarded_body(gateway, _HAIKU_4_5, client_body) + assert body == expected, _diff(expected, body) + + +def test_empty_thinking_block_is_stripped_and_redacted_thinking_is_kept(gateway: Gateway) -> None: + redacted: Final = {"type": "redacted_thinking", "data": "EmwKAhgBEgy3va3pzix"} + client_body: Final = _history_body( + ({"type": "thinking", "thinking": "", "signature": "EqQBCkgIBRABGAIiQM"}, redacted, _TOOL_CALL) + ) + expected: Final = {**_with_assistant_content(client_body, (redacted, _TOOL_CALL)), "model": _HAIKU_4_5} + body: Final = _forwarded_body(gateway, _HAIKU_4_5, client_body) + assert body == expected, _diff(expected, body) + + +@pytest.mark.parametrize( + ("upstream_model", "reasoning_effort"), + ( + pytest.param(_HAIKU_4_5, "turbo", id="unknown-value"), + pytest.param(_OPUS_4_6, "xhigh", id="level-the-model-lacks"), + ), +) +def test_unsupported_reasoning_effort_is_rejected_before_reaching_anthropic( + gateway: Gateway, upstream_model: str, reasoning_effort: str +) -> None: + with wire_server(lambda request: Reply()) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{upstream_model}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY + ) + response: Final = gateway.request( + "POST", "/v1/messages", {**_client_body(reasoning_effort=reasoning_effort), "model": model} + ) + assert response.status_code == 400, response.text + assert response.json()["error"]["type"] == "invalid_request_error", response.text + assert wire.drain() == () diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_response_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_response_wire.py new file mode 100644 index 00000000000..bd2659b6242 --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_response_wire.py @@ -0,0 +1,137 @@ +import json +import uuid +from typing import Final + +from integration._support import claude_code as cc +from integration._support.client import Gateway +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +_MODEL: Final = "claude-haiku-4-5" +_THINKING: Final = "the user wants a single word" +_SIGNATURE: Final = "EqQBCkgIBRABGAIiQLz" +_REDACTED: Final = "EmwKAhgBEgy3va3pzixlit" + + +def _reasoning_stream(identity: str) -> tuple[bytes, ...]: + return ( + cc.sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": identity, + "type": "message", + "role": "assistant", + "model": _MODEL, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 12, "output_tokens": 1}, + }, + }, + ), + cc.sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + }, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": _THINKING}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": _SIGNATURE}}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + cc.sse_frame( + "content_block_start", + { + "type": "content_block_start", + "index": 1, + "content_block": {"type": "redacted_thinking", "data": _REDACTED}, + }, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 1}), + cc.sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 2, "content_block": {"type": "text", "text": ""}}, + ), + cc.sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 2, "delta": {"type": "text_delta", "text": "PONG"}}, + ), + cc.sse_frame("content_block_stop", {"type": "content_block_stop", "index": 2}), + cc.sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 30}, + }, + ), + cc.sse_frame("message_stop", {"type": "message_stop"}), + ) + + +def _reasoning_message(identity: str) -> dict[str, JsonValue]: + return { + "id": identity, + "type": "message", + "role": "assistant", + "model": _MODEL, + "content": [ + {"type": "thinking", "thinking": _THINKING, "signature": _SIGNATURE}, + {"type": "redacted_thinking", "data": _REDACTED}, + {"type": "text", "text": "PONG"}, + ], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 12, "output_tokens": 30}, + } + + +def _content_events(stream: str) -> tuple[tuple[str, dict[str, object]], ...]: + return tuple(event for event in cc.sse_events(stream) if event[0].startswith("content_block_")) + + +def _client_body(stream: bool) -> dict[str, JsonValue]: + return { + **cc.claude_code_request(f"cache-bust-{uuid.uuid4().hex}"), + "thinking": {"type": "enabled", "budget_tokens": 2048}, + "stream": stream, + } + + +def test_streamed_thinking_signature_and_redacted_thinking_reach_the_client_unchanged(gateway: Gateway) -> None: + identity: Final = f"msg_{uuid.uuid4().hex}" + upstream_frames: Final = _reasoning_stream(identity) + + def respond(request: Request) -> Reply: + return Reply(chunks=upstream_frames, content_type="text/event-stream") + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request("POST", "/v1/messages", {**_client_body(stream=True), "model": model}) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 + assert _content_events(response.text) == _content_events(b"".join(upstream_frames).decode()), response.text + + +def test_non_streamed_thinking_signature_and_redacted_thinking_reach_the_client_unchanged(gateway: Gateway) -> None: + identity: Final = f"msg_{uuid.uuid4().hex}" + upstream_message: Final = _reasoning_message(identity) + + def respond(request: Request) -> Reply: + return Reply(body=json.dumps(upstream_message).encode()) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{_MODEL}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + response: Final = gateway.request("POST", "/v1/messages", {**_client_body(stream=False), "model": model}) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 + assert response.json()["content"] == upstream_message["content"], response.text diff --git a/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_token_pricing_wire.py b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_token_pricing_wire.py new file mode 100644 index 00000000000..ed12e0216ef --- /dev/null +++ b/tests/integration/messages_endpoint/providers/anthropic/reasoning/test_anthropic_reasoning_token_pricing_wire.py @@ -0,0 +1,71 @@ +import json +import uuid +from typing import Final + +import pytest +from integration._support import claude_code as cc +from integration._support.client import Gateway, eventually +from integration._support.database import read_rows +from integration._support.wire import Reply, Request, wire_server + +_MODEL: Final = "claude-haiku-4-5" +_INPUT_RATE: Final = 1e-6 +_OUTPUT_RATE: Final = 2e-6 +_REASONING_RATE: Final = 7e-6 + + +def test_reported_thinking_tokens_are_billed_at_the_reasoning_rate_and_the_rest_at_the_output_rate( + gateway: Gateway, +) -> None: + identity: Final = f"msg_{uuid.uuid4().hex}" + + def respond(request: Request) -> Reply: + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": _MODEL, + "content": [ + {"type": "thinking", "thinking": "count the words", "signature": "EqQBCkgIBRABGAIiQLz"}, + {"type": "text", "text": "PONG"}, + ], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 100, + "output_tokens": 50, + "output_tokens_details": {"thinking_tokens": 30}, + }, + } + ).encode() + ) + + with wire_server(respond) as wire, gateway.scenario() as scenario: + model: Final = scenario.model( + model=f"anthropic/{_MODEL}", + api_base=wire.url, + api_key=cc.ANTHROPIC_API_KEY, + input_cost_per_token=_INPUT_RATE, + output_cost_per_token=_OUTPUT_RATE, + output_cost_per_reasoning_token=_REASONING_RATE, + ) + response: Final = gateway.request( + "POST", + "/v1/messages", + { + **cc.claude_code_request(f"cache-bust-{uuid.uuid4().hex}"), + "thinking": {"type": "enabled", "budget_tokens": 2048}, + "stream": False, + "model": model, + }, + ) + assert response.status_code == 200, response.text + assert len(wire.drain()) == 1 + rows: Final = eventually( + lambda: read_rows('SELECT spend FROM "LiteLLM_SpendLogs" WHERE request_id=%s', (identity,)), + lambda values: len(values) == 1, + seconds=70, + ) + assert float(rows[0]["spend"]) == pytest.approx(100 * _INPUT_RATE + 20 * _OUTPUT_RATE + 30 * _REASONING_RATE)