scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

feat(auto_router): add a shunt preset that bounds large reads and delegates codegen

Browse source 
Ports the shunt technique for Claude Code
(https://engineering.atspotify.com/2026/9/portal-by-spotify-cut-my-claude-code-token-usage-by-90)
into the auto router, server-side, so it works for any client with no plugin to install.

shunt intercepts large file reads and boilerplate generation at the client with PreToolUse
hooks and hands them to a cheap worker model. This makes the same decision on the proxy: a
new always-on ShuntGuardrail injects bulk_read and code_write tool definitions pre-call, then
post-call rewrites large Read/Bash/bulk_read/code_write tool_use blocks into a Bash command
carrying shunt's own wc -l conditional, so small files still get read directly and only large
ones are delegated. Two new endpoints, /v1/bulk_read and /v1/code_write, run the worker call
through llm_router.acompletion with shunt's verbatim system prompts, so worker spend is
tracked against the calling key and team like any other request.

The guardrail arms per request off an auto-router marker's own litellm_params
(auto_router_shunt_min_lines and the two worker-model fields), the same shape
auto_router_compression already uses, so no guardrails: config entry is needed. It stays
inert when the resolved model has no shunt config.

In the UI, "Shunt" is one entry in the existing preset dropdown. Its tiers resolve through
fallback chains against whatever models the proxy actually has, so it greys out only when
there are no chat models at all, never merely because the named models are absent, and all
three settings stay editable under Advanced.

Streaming buffers the whole response before rewriting, matching tool_permission.py, because
input_json_delta fragments split mid-token. An unparseable stream passes through untouched:
shunt is an optimization, not a safety control, so a request should never fail because the
rewrite could not run.

The CLI's Bash allow rules only reduce prompts for the generated commands. Claude Code
matches rule text before the first wildcard with no host-aware matching, so they cannot pin
the destination, and the docs recommend a PreToolUse hook where a real boundary is needed.
This commit is contained in:
moe-berri 2026-09-07 13:37:21 -07:00
parent b618c7ad86
commit 56009d32dd
27 changed files with 2429 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/proxy/_types.py
View file

@ -489,6 +489,10 @@ class LiteLLMRoutes(enum.Enum):
"/v1/rag/ingest",
"/rag/query",
"/v1/rag/query",
"/bulk_read",
"/v1/bulk_read",
"/code_write",
"/v1/code_write",
]
anthropic_routes = [

12
litellm/proxy/client/cli/commands/autoroute/commands.py
View file

@ -37,7 +37,7 @@ from .process import (
terminate,
write_pid_record,
)
from .settings import merge_claude_settings_static_token
from .settings import merge_claude_settings_shunt_permissions, merge_claude_settings_static_token
from .wizard import run_configure_wizard
_GENERATED_CONFIG_ADAPTER: Final = TypeAdapter(dict[str, JsonValue])
@ -156,10 +156,16 @@ def up(port: int) -> None:
ClaudeBackupRecord(existed=original_existed, content=original_settings if original_existed else None),
AUTOROUTE_BACKUP_PATH,
)
merged: Final = merge_claude_settings_static_token(original_settings, base_url, master_key)
with_token: Final = merge_claude_settings_static_token(original_settings, base_url, master_key)
# Applied unconditionally: an unused allow rule matches no real command and has no
# runtime effect, so this stays correct whether or not the running config actually
# arms shunt (`configure` has no shunt option yet; this is ready for when it does).
merged: Final = merge_claude_settings_shunt_permissions(with_token)
CLAUDE_SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
with secure_create(CLAUDE_SETTINGS_PATH) as f:
json.dump(merged, f, indent=2)
# merged nests MappingProxyType at every level (see merge_claude_settings_shunt_
# permissions); default=dict is what the JSON encoder needs to serialize those.
json.dump(merged, f, indent=2, default=dict)
except ClaudeSettingsError as e:
terminate(process.pid)
clear_pid_record()

45
litellm/proxy/client/cli/commands/autoroute/settings.py
View file

@ -1,3 +1,5 @@
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from pydantic import JsonValue
@ -5,6 +7,8 @@ from pydantic import JsonValue
from .config import AUTOROUTER_MODEL_NAME
ENV_KEY: Final = "env"
PERMISSIONS_KEY: Final = "permissions"
ALLOW_KEY: Final = "allow"
API_KEY_HELPER_KEY: Final = "apiKeyHelper"
ANTHROPIC_API_KEY_KEY: Final = "ANTHROPIC_API_KEY"
ANTHROPIC_AUTH_TOKEN_KEY: Final = "ANTHROPIC_AUTH_TOKEN"
@ -22,6 +26,20 @@ ANTHROPIC_DEFAULT_MODEL_ENV_KEYS: Final = (
"ANTHROPIC_DEFAULT_OPUS_MODEL",
)
# Reduces prompts for the curl commands the shunt guardrail's Read/Bash rewrite generates
# (litellm/proxy/guardrails/auto_router_shunt.py); it is NOT a security boundary. Claude Code
# matches an allow rule's text before its first "*" verbatim, with no argument- or host-aware
# matching, so a rule narrow enough to name "curl" at all cannot also pin the destination the
# trailing "*" is free to name any URL. A real boundary needs a PreToolUse hook instead, which
# is the docs' own recommendation for exactly this case.
SHUNT_BASH_ALLOW_RULES: Final = (
"Bash(curl -sS -F question=*)",
"Bash(curl -sS -F spec=*)",
)
_NO_MAPPING: Final[Mapping[str, JsonValue]] = MappingProxyType({})
_NO_RULES: Final[tuple[JsonValue, ...]] = ()
def merge_claude_settings_static_token(
settings: dict[str, JsonValue], base_url: str, auth_token: str
@ -48,4 +66,29 @@ def merge_claude_settings_static_token(
return merged
__all__ = ["merge_claude_settings_static_token"]
def merge_claude_settings_shunt_permissions(settings: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
"""Add the shunt allow rules to `permissions.allow`, preserving every other permissions key.
A naive top-level `{**settings, "permissions": {...}}` would replace the whole `permissions`
object, silently dropping any `deny`/`ask` rules the caller already has -- exactly the trap
`merge_claude_settings_static_token` avoids for `env` by merging that key explicitly instead
of replacing it. This does the same for `permissions.allow`: read what's there, add only the
two shunt rules if they're not already present, and leave `deny`/`ask`/anything else alone.
Returns a `MappingProxyType` nested at every level rather than a plain dict, so the caller's
`json.dump(..., default=dict)` is what converts it back to something the JSON encoder accepts
-- the one place a concrete mutable mapping is genuinely needed, kept out of this function.
"""
raw_permissions: Final = settings.get(PERMISSIONS_KEY, _NO_MAPPING)
base_permissions: Final = raw_permissions if isinstance(raw_permissions, Mapping) else _NO_MAPPING
raw_allow: Final = base_permissions.get(ALLOW_KEY, _NO_RULES)
base_allow: Final = raw_allow if isinstance(raw_allow, (list, tuple)) else _NO_RULES
new_allow: Final = (*base_allow, *(rule for rule in SHUNT_BASH_ALLOW_RULES if rule not in base_allow))
permissions: Final = MappingProxyType({**base_permissions, ALLOW_KEY: new_allow})
return MappingProxyType({**settings, PERMISSIONS_KEY: permissions})
__all__ = [ # mutable-ok: __all__ must be a list per Python convention
"merge_claude_settings_shunt_permissions",
"merge_claude_settings_static_token",
]

567
litellm/proxy/guardrails/auto_router_shunt.py Normal file
View file

@ -0,0 +1,567 @@
"""
Server-side port of Spotify's ``shunt`` Claude Code plugin
(https://engineering.atspotify.com/2026/9/portal-by-spotify-cut-my-claude-code-token-usage-by-90),
as an auto-router preset rather than a client-side plugin.
shunt intercepts large file reads and boilerplate generation at the client via Claude Code
``PreToolUse`` hooks and delegates them to a cheap worker model. This module does the same
decision on the proxy instead: it arms via ``auto_router_shunt_min_lines`` /
``auto_router_shunt_bulk_read_model`` / ``auto_router_shunt_code_write_model`` on an
auto-router marker deployment (the same "read `litellm_params` off the resolved deployment, no
``guardrails:`` config entry needed" shape as ``auto_router_compression.py``), then a single
always-on ``ShuntGuardrail`` callback injects ``bulk_read``/``code_write`` tool definitions
pre-call and rewrites large ``Read``/``Bash``/``bulk_read``/``code_write`` tool_use blocks
post-call so the file bytes and generated code never reach the routed model's context.
"""
import json
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Final
from litellm.integrations.custom_logger import CustomLogger
from litellm.router_utils.auto_router_model_naming import AUTO_ROUTER_MODEL_PREFIX
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, AsyncIterable
from litellm.caching.dual_cache import DualCache
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.guardrails.shunt_rewrite import ShuntBashRewrite
from litellm.router import Router
from litellm.types.utils import CallTypesLiteral, ModelResponseStream
@dataclass(frozen=True, slots=True)
class ShuntConfig:
"""An auto router's shunt settings, resolved from its marker deployment."""
min_lines: int
bulk_read_model: str
code_write_model: str
def _config_from_litellm_params(
litellm_params: Mapping[str, object], *, default_model: str | None
) -> ShuntConfig | None:
raw_min_lines: Final = litellm_params.get("auto_router_shunt_min_lines")
if not isinstance(raw_min_lines, int):
return None
fallback_model: Final = default_model or ""
raw_bulk_read: Final = litellm_params.get("auto_router_shunt_bulk_read_model")
raw_code_write: Final = litellm_params.get("auto_router_shunt_code_write_model")
return ShuntConfig(
min_lines=raw_min_lines,
bulk_read_model=raw_bulk_read if isinstance(raw_bulk_read, str) and raw_bulk_read else fallback_model,
code_write_model=raw_code_write if isinstance(raw_code_write, str) and raw_code_write else fallback_model,
)
def shunt_config_for_model(
llm_router: "Router | None",
model_alias: str,
team_id: str | None,
request_tags: Sequence[str],
) -> ShuntConfig | None:
"""The shunt config of the auto router marker `model_alias` resolves to, or None.
Mirrors `auto_router_compression.policy_for_model`'s tag-scoped resolution exactly, so an
alias with several tag-scoped markers cannot arm shunt under one and route under another.
"""
if llm_router is None:
return None
deployments: Final = llm_router.get_model_list(model_name=model_alias, team_id=team_id) or ()
markers: Final = tuple(
litellm_params
for deployment in deployments
if isinstance(litellm_params := deployment.get("litellm_params"), Mapping) # pyright: ignore[reportUnnecessaryIsInstance] # filters out non-Mapping
and str(litellm_params.get("model", "")).startswith(AUTO_ROUTER_MODEL_PREFIX)
)
requested: Final = frozenset(request_tags)
tag_matched: Final = tuple(
params for params in markers if (tags := params.get("tags")) and requested.issuperset(frozenset(tags))
)
# Untagged only: a marker scoped to tags this request lacks describes other traffic.
untagged: Final = tuple(params for params in markers if not params.get("tags"))
# Lazy, so the first marker carrying a config wins and the rest are never read.
candidates: Final = (_config_from_marker(params) for params in (*tag_matched, *untagged))
return next((config for config in candidates if config is not None), None)
def _config_from_marker(litellm_params: Mapping[str, object]) -> ShuntConfig | None:
default_model: Final = litellm_params.get("auto_router_default_model") or litellm_params.get(
"complexity_router_default_model"
)
return _config_from_litellm_params(
litellm_params, default_model=default_model if isinstance(default_model, str) else None
)
# Tool descriptions carry shunt's own SKILL.md guidance: this is shunt's skills layer,
# delivered as tool metadata instead of a bundled markdown file, since there is no client-side
# plugin here for a skill file to live in.
BULK_READ_TOOL_NAME: Final = "bulk_read"
CODE_WRITE_TOOL_NAME: Final = "code_write"
# Held as JSON source, not dict literals: these are JSON Schema documents that go straight into
# the outbound provider payload, so they must stay plain JSON-serializable dicts (a
# MappingProxyType raises in the JSON encoder). Parsing the schema from the notation it is
# written in keeps one construction site instead of a suppression on every nested literal.
_TOOL_DEFINITIONS_JSON: Final = """
{
"bulk_read": {
"description": "Delegate reading one or more large files to a cheaper model when you only need a summary or an answer to a specific question about their contents, not exact line-level content for editing. Each call is independent; ask a follow-up by calling again with the same paths.",
"input_schema": {
"type": "object",
"properties": {
"question": {"type": "string", "description": "What to find out from the files."},
"paths": {"type": "array", "items": {"type": "string"}, "description": "File paths to read."}
},
"required": ["question", "paths"]
}
},
"code_write": {
"description": "Delegate generating boilerplate code (tests, config, stubs, or anything more than 80% predictable from a reference file) to a cheaper model. Match the reference file's patterns, conventions, naming, and style exactly.",
"input_schema": {
"type": "object",
"properties": {
"spec": {"type": "string", "description": "What to generate."},
"reference": {"type": "string", "description": "Path to a file whose patterns the output should match."},
"target": {"type": "string", "description": "Optional path to write the generated code to."}
},
"required": ["spec", "reference"]
}
}
}
"""
def _anthropic_tool(name: str) -> Mapping[str, object]:
"""The Anthropic-shape tool definition for `name`, as a fresh plain dict.
Fresh per call, never a shared module-level dict: these go into the outbound payload, and
handing every request the same mutable object would let one request's downstream mutation
(a provider transform normalizing a schema in place, say) leak into every later request.
"""
definition: Final = json.loads(_TOOL_DEFINITIONS_JSON)[name]
return {"name": name, **definition} # mutable-ok: goes into the outbound provider payload as plain JSON
def _as_openai_function_tool(anthropic_tool: Mapping[str, object]) -> Mapping[str, object]:
"""The OpenAI `{"type": "function", "function": {...}}` shape of an Anthropic-style tool."""
return { # mutable-ok: goes into the outbound provider payload as plain JSON
"type": "function",
"function": { # mutable-ok: same
"name": anthropic_tool["name"],
"description": anthropic_tool["description"],
"parameters": anthropic_tool["input_schema"],
},
}
_SHUNT_CALL_TYPES: Final = frozenset({"completion", "acompletion", "anthropic_messages", "aanthropic_messages"})
def _request_base_url(data: Mapping[str, object]) -> str | None:
"""The scheme+host+port the client actually used to reach this proxy, or None.
Read from `data["proxy_server_request"]["url"]` (the real request URL, stamped by
`litellm_pre_call_utils.py` before routing) rather than an env var or a hardcoded
localhost, so the generated curl reaches the proxy on whatever host/port/scheme the
caller is actually using.
"""
proxy_server_request: Final = data.get("proxy_server_request")
if not isinstance(proxy_server_request, Mapping):
return None
raw_url: Final = proxy_server_request.get("url")
if not isinstance(raw_url, str) or not raw_url:
return None
from urllib.parse import urlsplit
parts: Final = urlsplit(raw_url)
if not parts.scheme or not parts.netloc:
return None
return f"{parts.scheme}://{parts.netloc}"
def _request_auth_header(data: Mapping[str, object]) -> str | None:
"""The client's own `authorization` header, or None.
Forwarded rather than a stored key: the generated curl authenticates to this proxy as the
same caller who sent the request, so it is billed and rate-limited the same way, and this
module never needs to hold or mint a credential of its own.
"""
secret_fields: Final = data.get("secret_fields")
if not isinstance(secret_fields, Mapping):
return None
raw_headers: Final = secret_fields.get("raw_headers")
if not isinstance(raw_headers, Mapping):
return None
header: Final = raw_headers.get("authorization")
return header if isinstance(header, str) and header else None
def _resolve_shunt_config(data: Mapping[str, object]) -> ShuntConfig | None:
"""The armed `ShuntConfig` for this request's resolved model, or None.
Shared by both hooks so pre-call tool injection and post-call rewriting agree on whether
shunt is armed for this request; `data["model"]` is reliably the caller-facing alias at
both points, since the router unpacks it into a fresh local `kwargs` at the call site
rather than mutating the request dict this guardrail was handed.
"""
model: Final = data.get("model")
if not isinstance(model, str) or not model:
return None
from litellm.proxy.guardrails.auto_router_compression import team_id_from_request
from litellm.proxy.proxy_server import llm_router
from litellm.router_strategy.tag_based_routing import (
_get_tags_from_request_kwargs, # pyright: ignore[reportPrivateUsage] # same helper router.py and auto_router_compression.py already use
)
return shunt_config_for_model(
llm_router=llm_router,
model_alias=model,
team_id=team_id_from_request(data),
request_tags=_get_tags_from_request_kwargs(data),
)
DEFAULT_BULK_READ_QUESTION: Final = "Summarize this file's exports and overall structure."
@dataclass(frozen=True, slots=True)
class _ShuntEndpoints:
bulk_read_url: str
code_write_url: str
auth_header: str
def _endpoints_for_request(data: Mapping[str, object], model_alias: str) -> "_ShuntEndpoints | None":
"""Where this request's generated curl commands should point, or None if unreachable.
None when the base URL or the caller's own auth header can't be recovered: without both,
a generated command could not reach this proxy as this caller, so the tool_use is left
unmodified rather than shipped with a broken command.
"""
base_url: Final = _request_base_url(data)
auth_header: Final = _request_auth_header(data)
if base_url is None or auth_header is None:
return None
from urllib.parse import quote
router_query: Final = f"router={quote(model_alias)}"
return _ShuntEndpoints(
bulk_read_url=f"{base_url}/v1/bulk_read?{router_query}",
code_write_url=f"{base_url}/v1/code_write?{router_query}",
auth_header=auth_header,
)
def _string_list(value: object) -> tuple[str, ...] | None:
if not isinstance(value, list):
return None
strings: Final = tuple(item for item in value if isinstance(item, str))
return strings if len(strings) == len(value) else None
def _bash_replacement_for_tool_use(
name: str, tool_input: Mapping[str, object], config: ShuntConfig, endpoints: _ShuntEndpoints
) -> "ShuntBashRewrite | None":
"""The `Bash` command a tool_use becomes, or None to leave it untouched.
Four sources, all converging on the same generated `curl`: the model's own explicit
`bulk_read`/`code_write` call, or an untargeted `Read`/bare `cat`/`head`/`tail`/`less`/
`more` call this guardrail bounds on the model's behalf.
"""
from litellm.proxy.guardrails.shunt_rewrite import (
build_bounded_read_command,
build_bulk_read_command,
build_code_write_command,
extract_bare_read_path,
is_targeted_read,
)
if name == BULK_READ_TOOL_NAME:
question: Final = tool_input.get("question")
paths: Final = _string_list(tool_input.get("paths"))
if not isinstance(question, str) or not question or not paths:
return None
return build_bulk_read_command(
question=question,
paths=paths,
bulk_read_endpoint=endpoints.bulk_read_url,
auth_header=endpoints.auth_header,
)
if name == CODE_WRITE_TOOL_NAME:
spec: Final = tool_input.get("spec")
reference: Final = tool_input.get("reference")
if not isinstance(spec, str) or not spec or not isinstance(reference, str) or not reference:
return None
target: Final = tool_input.get("target")
return build_code_write_command(
spec=spec,
reference=reference,
target=target if isinstance(target, str) and target else None,
code_write_endpoint=endpoints.code_write_url,
auth_header=endpoints.auth_header,
)
if name == "Read":
path: Final = tool_input.get("file_path")
if not isinstance(path, str) or not path:
return None
if is_targeted_read(tool_input.get("offset"), tool_input.get("limit")):
return None
return build_bounded_read_command(
path=path,
question=DEFAULT_BULK_READ_QUESTION,
min_lines=config.min_lines,
bulk_read_endpoint=endpoints.bulk_read_url,
auth_header=endpoints.auth_header,
)
if name == "Bash":
command: Final = tool_input.get("command")
if not isinstance(command, str) or not command:
return None
bare_path: Final = extract_bare_read_path(command)
if bare_path is None:
return None
return build_bounded_read_command(
path=bare_path,
question=DEFAULT_BULK_READ_QUESTION,
min_lines=config.min_lines,
bulk_read_endpoint=endpoints.bulk_read_url,
auth_header=endpoints.auth_header,
)
return None
def _rewrite_openai_tool_call(tool_call: object, config: ShuntConfig, endpoints: _ShuntEndpoints) -> bool:
"""Rewrite one OpenAI-shape tool call's `function` in place. Returns whether it changed.
A custom tool call (`ChatCompletionMessageCustomToolCall`) has no `.function` attribute at
all, so it is never a shunt-shaped call and is skipped rather than inspected. The return
value lets a caller streaming the response skip re-serializing a stream nothing in it
changed, matching `tool_permission.py`'s own "an allowed stream must not be re-serialized".
"""
from litellm.types.utils import ChatCompletionMessageToolCall, Function
if not isinstance(tool_call, ChatCompletionMessageToolCall):
return False
name: Final = tool_call.function.name
if not isinstance(name, str):
return False
try:
tool_input: Final = json.loads(tool_call.function.arguments or "{}")
except (json.JSONDecodeError, TypeError):
return False
if not isinstance(tool_input, Mapping):
return False
rewrite: Final = _bash_replacement_for_tool_use(name, tool_input, config, endpoints)
if rewrite is None:
return False
tool_call.function = Function( # rebind-ok: rewrites the provider response in place, mirrors tool_permission.py
name="Bash",
arguments=json.dumps({"command": rewrite.command}), # mutable-ok: serialized immediately, never held
)
return True
def _rewrite_anthropic_content_block(block: object, config: ShuntConfig, endpoints: _ShuntEndpoints) -> bool:
"""Rewrite one Anthropic-shape content block's `name`/`input` in place. Returns whether it changed."""
if not isinstance(block, dict) or block.get("type") != "tool_use":
return False
name: Final = block.get("name")
tool_input: Final = block.get("input")
if not isinstance(name, str) or not isinstance(tool_input, Mapping):
return False
rewrite: Final = _bash_replacement_for_tool_use(name, tool_input, config, endpoints)
if rewrite is None:
return False
block["name"] = "Bash" # rebind-ok: rewrites the provider response in place, mirrors tool_permission.py
block["input"] = {"command": rewrite.command} # rebind-ok: same # mutable-ok: part of the response payload
return True
def _rewrite_openai_response_in_place(response: object, config: ShuntConfig, endpoints: _ShuntEndpoints) -> bool:
"""Rewrite every shunt-shaped tool call across all choices. Returns whether any changed.
Built as a tuple comprehension, not a short-circuiting `any(...)`: every tool call must be
inspected and rewritten regardless of whether an earlier one already changed, so evaluation
can't stop at the first True the way `any` would.
"""
from litellm.types.utils import Choices
results: Final = tuple(
_rewrite_openai_tool_call(tool_call, config, endpoints)
for choice in getattr(response, "choices", ())
if isinstance(choice, Choices)
for tool_call in choice.message.tool_calls or ()
)
return any(results)
def _rewrite_anthropic_response_in_place(
response: dict, # mutable-ok: the provider response this rewrites in place, as tool_permission.py does
config: ShuntConfig,
endpoints: _ShuntEndpoints,
) -> bool:
"""Rewrite every shunt-shaped tool_use block in `response["content"]`. Returns whether any changed.
Same full-evaluation shape as `_rewrite_openai_response_in_place`, for the same reason.
"""
content: Final = response.get("content")
if not isinstance(content, list):
return False
results: Final = tuple(_rewrite_anthropic_content_block(block, config, endpoints) for block in content)
return any(results)
def _tools_payload(
existing: Sequence[object], added: Sequence[Mapping[str, object]]
) -> list[object]: # mutable-ok: outbound provider payload
"""`existing` plus `added`, as the plain list the outbound provider payload must hold."""
return [*existing, *added] # mutable-ok: outbound provider payload; transforms downstream append to it
class ShuntGuardrail(CustomLogger):
"""Always-registered callback (see the module docstring) that arms per-request from an
auto-router marker's `litellm_params`, never from a `guardrails:` config entry.
"""
async def async_pre_call_hook(
self,
user_api_key_dict: "UserAPIKeyAuth",
cache: "DualCache",
data: dict, # mutable-ok: CustomLogger declares `data: dict`; narrowing breaks the override
call_type: "CallTypesLiteral",
) -> dict | None: # mutable-ok: the base class's own return type; the caller re-reads this dict
if call_type not in _SHUNT_CALL_TYPES:
return None
config: Final = _resolve_shunt_config(data)
if config is None:
return None
use_anthropic_format: Final = call_type in ("anthropic_messages", "aanthropic_messages")
anthropic_tools: Final = (_anthropic_tool(BULK_READ_TOOL_NAME), _anthropic_tool(CODE_WRITE_TOOL_NAME))
new_tools: Final = (
anthropic_tools if use_anthropic_format else tuple(_as_openai_function_tool(t) for t in anthropic_tools)
)
existing_tools: Final = data.get("tools")
# Mutating `data` is how this hook is documented to modify a request; litellm_skills/
# main.py does the same `data["tools"] = ...` assignment.
data["tools"] = _tools_payload( # rebind-ok: the hook's documented way to modify a request
existing_tools if isinstance(existing_tools, list) else (), new_tools
)
return data
async def async_post_call_success_hook(
self,
data: dict, # mutable-ok: CustomLogger declares `data: dict`; narrowing breaks the override
user_api_key_dict: "UserAPIKeyAuth",
response: object,
) -> object:
from litellm.types.utils import ModelResponse
config: Final = _resolve_shunt_config(data)
if config is None:
return response
model: Final = data.get("model")
endpoints: Final = _endpoints_for_request(data, model) if isinstance(model, str) and model else None
if endpoints is None:
return response
if isinstance(response, ModelResponse):
_rewrite_openai_response_in_place(response, config, endpoints)
return response
if isinstance(response, dict):
_rewrite_anthropic_response_in_place(response, config, endpoints)
return response
return response
async def async_post_call_streaming_iterator_hook(
self,
user_api_key_dict: "UserAPIKeyAuth",
response: "AsyncIterable[ModelResponseStream]",
request_data: dict, # mutable-ok: CustomLogger declares `request_data: dict`; narrowing breaks the override
) -> "AsyncGenerator[ModelResponseStream, None]":
"""Buffer the whole stream, rewrite any shunt-shaped tool_use, then replay it.
Buffer-then-replay, not per-fragment rewriting, matching `tool_permission.py`'s own
streaming hook: `input_json_delta` fragments split mid-token (confirmed against a real
Anthropic trace during design), so a tool_use's `input` can only be read once its
`content_block_stop` has arrived. Unlike that guardrail, an unparseable or unassemblable
stream is passed through unmodified rather than raised on: shunt is an optimization, not
a safety control, so a request should never fail because this rewrite couldn't run.
The declared return type matches the base class and `tool_permission.py`'s own override:
on the Anthropic path this actually yields raw `bytes` SSE frames, not
`ModelResponseStream` objects, the same documented mismatch `tool_permission.py` carries.
"""
from litellm.main import stream_chunk_builder
from litellm.proxy.guardrails.anthropic_sse import (
anthropic_sse_chunks_from_response,
assemble_anthropic_sse_stream,
is_raw_sse_stream,
)
from litellm.types.utils import ModelResponse, TextCompletionResponse
# Declared element type matches `tool_permission.py`'s own override rather than the
# `object` the raw-SSE path really carries: see the docstring's note on that mismatch.
all_chunks: Final[
list[ModelResponseStream]
] = [ # mutable-ok: stream_chunk_builder/is_raw_sse_stream both take a concrete list
chunk async for chunk in response
]
config: Final = _resolve_shunt_config(request_data)
model: Final = request_data.get("model")
endpoints: Final = (
_endpoints_for_request(request_data, model)
if config is not None and isinstance(model, str) and model
else None
)
if config is None or endpoints is None:
for chunk in all_chunks:
yield chunk
return
if is_raw_sse_stream(all_chunks):
assembled: Final = assemble_anthropic_sse_stream(all_chunks)
if assembled is None:
# Unparseable: pass the stream through exactly as received rather than fail the
# request over an optimization that could not run.
for chunk in all_chunks:
yield chunk
return
changed: Final = _rewrite_openai_response_in_place(assembled, config, endpoints)
if not changed:
for chunk in all_chunks:
yield chunk
return
for sse_chunk in anthropic_sse_chunks_from_response(assembled):
yield sse_chunk
return
assembled_openai: Final[ModelResponse | TextCompletionResponse | None] = stream_chunk_builder(chunks=all_chunks)
if not isinstance(assembled_openai, ModelResponse):
for chunk in all_chunks:
yield chunk
return
changed_openai: Final = _rewrite_openai_response_in_place(assembled_openai, config, endpoints)
if not changed_openai:
for chunk in all_chunks:
yield chunk
return
from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
async for chunk in MockResponseIterator(model_response=assembled_openai):
yield chunk

144
litellm/proxy/guardrails/shunt_rewrite.py Normal file
View file

@ -0,0 +1,144 @@
"""
Decides which tool_use blocks shunt rewrites, and builds the Bash command that replaces them.
Ports two decisions from Spotify's shunt plugin (see auto_router_shunt.py's module docstring):
`check-file-size`'s "is this an untargeted Read on a large file" gate, and
`check-bash-read`'s "is this cat/head/tail/less/more on a bare file path" parser, including its
documented parser bug (an option-value token like the `5` in `head -n 5 file` is mistaken for
the path). shunt's version tolerates that bug because its hook runs on the file's own machine
and checks the guessed path exists before blocking; a misparse then falls through to `allow`,
leaving the original command untouched. This module has no filesystem access to make that same
check, so it must never rewrite a command it isn't sure it parsed correctly — see
`extract_bare_read_path`'s docstring for how the port preserves shunt's fail-safe direction.
"""
import re
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Final
_READ_COMMANDS: Final = ("cat", "head", "tail", "less", "more")
_READ_COMMAND_PATTERN: Final = re.compile(r"^(?:" + "|".join(_READ_COMMANDS) + r")\s")
_PIPE_OR_REDIRECT_PATTERN: Final = re.compile(r"[|>]")
def is_targeted_read(offset: object, limit: object) -> bool:
"""Whether a `Read`-shaped tool call already names a bounded section.
Mirrors `check-file-size`'s "offset or limit set" gate, including its documented bypass:
`offset: 0` or `limit: 0` count as targeted, matching shunt's own behavior exactly.
"""
return offset is not None or limit is not None
def extract_bare_read_path(command: str) -> str | None:
"""The file path a bare `cat`/`head`/`tail`/`less`/`more` command reads, or None.
None means "do not rewrite this command": either it isn't a bulk read (piped, redirected,
not one of the five commands), or no non-flag argument was found.
shunt's own parser walks flags as if none of them take a value, which misreads a flag's
value token as the path — e.g. the `5` in `head -n 5 file`. shunt tolerates this because its
hook runs on the file's own machine and checks the guessed path actually exists before
blocking; a nonexistent guess falls through to `allow`, leaving the original command
untouched. This port has no filesystem to make that same check, so it cannot rely on a
downstream existence check to catch a bad guess — treating a wrong guess as the real path
would rewrite the command against a file the model never named. A bare numeric token (as
every flag value in the five commands' own option sets happens to be — `-n`, `head -c`,
`tail -n +N`) is skipped as a likely flag value rather than returned, which fixes shunt's
documented bug for exactly the case it names rather than reproducing it. A path made only
of digits (or `tail`'s `+N` follow-from-line syntax) is stripped either way, matching
shunt's own scope: neither version claims to handle a bare-numeric filename correctly.
Like shunt's own bash word-splitting, this does not handle a quoted path containing spaces
(`cat "my file.txt"` returns `"my`) — matching parity, not a regression.
"""
if _PIPE_OR_REDIRECT_PATTERN.search(command):
return None
match: Final = _READ_COMMAND_PATTERN.match(command)
if match is None:
return None
args: Final = command[match.end() :].split()
for arg in args:
if arg.startswith("-") or arg.lstrip("+-").isdigit():
continue
return arg.strip("\"'")
return None
@dataclass(frozen=True, slots=True)
class ShuntBashRewrite:
"""A Bash command that replaces an intercepted tool_use, and why."""
command: str
note: str
def _quote(value: str) -> str:
return value.replace('"', '\\"')
def build_bounded_read_command(
*, path: str, question: str, min_lines: int, bulk_read_endpoint: str, auth_header: str
) -> ShuntBashRewrite:
"""The shunt conditional: read small files directly, delegate large ones.
One template for both rewrite sources (an untargeted `Read` tool call, and a bare
`cat`/`head`/`tail`/`less`/`more` command) since both made the identical decision in shunt.
`-sS` on curl mirrors shunt's own `-s` plus fails loudly on a server error rather than
silently returning an HTML error page as if it were the answer.
"""
quoted_path: Final = _quote(path)
quoted_question: Final = _quote(question)
command: Final = (
f'L=$(wc -l < "{quoted_path}" 2>/dev/null || echo 0); '
f'if [ "$L" -gt {min_lines} ]; then '
f'echo "[shunt] {quoted_path}: $L lines, bounded read delegated" >&2; '
f'curl -sS -F question="{quoted_question}" -F "paths[]=@{quoted_path}" '
f'-H "Authorization: {auth_header}" "{bulk_read_endpoint}"; '
f'else cat "{quoted_path}"; fi'
)
return ShuntBashRewrite(
command=command,
note=f"Bounded read: files over {min_lines} lines are delegated to a cheaper model.",
)
def build_bulk_read_command(
*, question: str, paths: Sequence[str], bulk_read_endpoint: str, auth_header: str
) -> ShuntBashRewrite:
"""The curl a model's own explicit `bulk_read(question, paths)` tool call becomes.
Unconditional (no size check): the model chose to delegate, unlike the automatic bounding
`build_bounded_read_command` applies to a plain `Read`/`cat`/`head`/`tail` call.
"""
quoted_question: Final = _quote(question)
path_flags: Final = " ".join(f'-F "paths[]=@{_quote(path)}"' for path in paths)
command: Final = (
f'curl -sS -F question="{quoted_question}" {path_flags} '
f'-H "Authorization: {auth_header}" "{bulk_read_endpoint}"'
)
return ShuntBashRewrite(command=command, note="Delegated to a cheaper model via bulk_read.")
def build_code_write_command(
*, spec: str, reference: str, target: str | None, code_write_endpoint: str, auth_header: str
) -> ShuntBashRewrite:
"""The curl a model's own explicit `code_write(spec, reference, target)` tool call becomes.
shunt's own script writes straight to disk since it runs on the file's own machine; this
has no local filesystem, so the generated code returns to the client, which writes it
itself when `target` is given. Either way the generated code enters the client's context
only as a file write, never as text the routed model has to hold or repeat.
"""
quoted_spec: Final = _quote(spec)
quoted_reference: Final = _quote(reference)
request: Final = (
f'curl -sS -F spec="{quoted_spec}" -F "reference=@{quoted_reference}" '
f'-H "Authorization: {auth_header}" "{code_write_endpoint}"'
)
if target is None:
return ShuntBashRewrite(command=request, note="Delegated to a cheaper model via code_write.")
quoted_target: Final = _quote(target)
command: Final = f'{request} > "{quoted_target}"'
return ShuntBashRewrite(command=command, note=f"Delegated to a cheaper model via code_write, written to {target}.")

6
litellm/proxy/proxy_server.py
View file

@ -435,6 +435,7 @@ from litellm.proxy.discovery_endpoints import ui_discovery_endpoints_router
from litellm.proxy.fine_tuning_endpoints.endpoints import router as fine_tuning_router
from litellm.proxy.fine_tuning_endpoints.endpoints import set_fine_tuning_config
from litellm.proxy.google_endpoints.endpoints import router as google_router
from litellm.proxy.guardrails.auto_router_shunt import ShuntGuardrail
from litellm.proxy.guardrails.init_guardrails import (
init_guardrails_v2,
initialize_guardrails,
@ -637,6 +638,7 @@ from litellm.proxy.rerank_endpoints.endpoints import router as rerank_router
from litellm.proxy.response_api_endpoints.endpoints import router as response_router
from litellm.proxy.route_llm_request import route_request
from litellm.proxy.search_endpoints.endpoints import router as search_router
from litellm.proxy.shunt_endpoints.endpoints import router as shunt_router
from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager
from litellm.proxy.spend_tracking.budget_reservation import get_budget_window_start
from litellm.proxy.spend_tracking.spend_management_endpoints import (
@ -2304,6 +2306,9 @@ spend_counter_cache: Final = DualCache(default_in_memory_ttl=UserAPIKeyCacheTTLE
cli_sso_session_cache: Final = DualCache(default_in_memory_ttl=CLI_SSO_SESSION_TTL_SECONDS)
model_max_budget_limiter: Final = _PROXY_VirtualKeyModelMaxBudgetLimiter(dual_cache=spend_counter_cache)
litellm.logging_callback_manager.add_litellm_callback(model_max_budget_limiter)
# Always registered, no `guardrails:` config entry needed: arms per-request from an
# auto-router marker's own litellm_params. See auto_router_shunt.py's module docstring.
litellm.logging_callback_manager.add_litellm_callback(ShuntGuardrail())
redis_usage_cache: RedisCache | None = None # redis cache used for tracking spend, tpm/rpm limits
polling_via_cache_enabled: Literal["all"] | list[str] | bool = False
native_background_mode: list[str] = [] # Models that should use native provider background mode instead of polling
@ -18332,6 +18337,7 @@ app.include_router(rag_router)
app.include_router(video_router)
app.include_router(container_router)
app.include_router(search_router)
app.include_router(shunt_router)
app.include_router(image_router)
app.include_router(fine_tuning_router)
app.include_router(credential_router)

20
litellm/proxy/public_endpoints/autorouter_presets.json
View file

@ -112,5 +112,25 @@
"modality_pin_override": false,
"deployment_affinity": true
}
},
"shunt": {
"label": "Shunt",
"description": "Bounds large file/shell reads and delegates bulk-read and boilerplate-codegen tool calls to a cheap model, based on the shunt technique for AI coding agents (https://engineering.atspotify.com/2026/9/portal-by-spotify-cut-my-claude-code-token-usage-by-90). Ported server-side: works with any client, no plugin install. classifier_type is heuristic so this never counts against the licensed heuristic_v2 limit.",
"complexity_router_config": {
"tiers": {
"SIMPLE": ["claude-haiku-4-5", "gpt-5.6-luna", "gemini-2.5-flash-lite", "deepseek-v4-flash"],
"MEDIUM": ["claude-sonnet-5", "gpt-5.6-terra", "gemini-3.1-flash-lite"],
"COMPLEX": ["claude-opus-5", "gpt-5.6-sol", "gemini-3.7-flash"],
"REASONING": ["claude-opus-5", "gpt-5.6-sol", "gemini-3.1-pro-preview"]
},
"classifier_type": "heuristic",
"escalation_keywords": ["LITELLM ESCALATE"],
"classification_mode": "every_request",
"session_affinity": false,
"modality_routing": false,
"modality_pin_override": false,
"deployment_affinity": true
},
"auto_router_shunt_min_lines": 350
}
}

5
litellm/proxy/shunt_endpoints/__init__.py Normal file
View file

@ -0,0 +1,5 @@
"""Shunt worker endpoints for LiteLLM Proxy."""
from litellm.proxy.shunt_endpoints.endpoints import router
__all__ = ["router"] # mutable-ok: __all__ must be a list per Python convention

181
litellm/proxy/shunt_endpoints/endpoints.py Normal file
View file

@ -0,0 +1,181 @@
"""
`/v1/bulk_read` and `/v1/code_write`: the worker endpoints shunt's generated Bash commands
call. See `auto_router_shunt.py`'s module docstring for the source this ports.
Each request names the auto-router marker it belongs to (`router=<model_alias>`, the same
value the client originally called), so the worker model is the one configured on that
marker's `auto_router_shunt_bulk_read_model` / `auto_router_shunt_code_write_model`, not a
model chosen by the caller. The call goes through `llm_router.acompletion`, not
`litellm.acompletion` directly, so worker-model spend is tracked and budgeted against the
caller's key/team exactly like any other request.
"""
from enum import Enum
from types import MappingProxyType
from typing import TYPE_CHECKING, Annotated, Final
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import ProxyErrorTypes, ProxyException, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.guardrails.auto_router_shunt import ShuntConfig, shunt_config_for_model
from litellm.proxy.shunt_endpoints.worker import (
BULK_READ_SYSTEM_PROMPT,
CODE_WRITE_SYSTEM_PROMPT,
WORKER_TEMPERATURE,
build_bulk_read_message,
build_code_write_message,
strip_code_fences,
)
from litellm.types.llms.openai import ChatCompletionSystemMessage, ChatCompletionUserMessage
if TYPE_CHECKING:
from litellm.router import Router
router: Final = APIRouter()
# Both endpoints are registered twice, at the `/v1`-prefixed path the generated commands call
# and at the bare path, matching how the rest of the proxy exposes its native routes.
_AUTH_DEPENDENCIES: Final = [Depends(user_api_key_auth)] # mutable-ok: FastAPI's `dependencies=` takes a list
_SHUNT_TAGS: Final[list[str | Enum]] = ["shunt"] # mutable-ok: FastAPI's `tags=` takes an invariant list
def _worker_config(model_alias: str, user_api_key_dict: UserAPIKeyAuth) -> tuple["Router", ShuntConfig]:
from litellm.proxy.proxy_server import llm_router
if llm_router is None:
raise ProxyException(
message="LLM Router not found", type=ProxyErrorTypes.internal_server_error, param=None, code=500
)
config: Final = shunt_config_for_model(
llm_router=llm_router,
model_alias=model_alias,
team_id=user_api_key_dict.team_id,
request_tags=(),
)
if config is None:
raise ProxyException(
message=f"'{model_alias}' is not a shunt-armed auto router",
type=ProxyErrorTypes.bad_request_error,
param="router",
code=400,
)
return llm_router, config
async def _worker_text(
llm_router: "Router", *, model: str, system_prompt: str, message: str, team_id: str | None, label: str
) -> str:
"""The worker model's reply text, or a 502 if it produced none.
One call site for both endpoints: they differ only in model, system prompt, and message, so
the `acompletion` shape (temperature, non-streaming, team metadata) lives here once.
"""
system: Final = ChatCompletionSystemMessage(role="system", content=system_prompt)
user: Final = ChatCompletionUserMessage(role="user", content=message)
response: Final = await llm_router.acompletion(
model=model,
messages=[system, user], # mutable-ok: acompletion's own signature takes a concrete list
temperature=WORKER_TEMPERATURE,
stream=False,
metadata={"user_api_key_team_id": team_id}, # mutable-ok: same
)
text: Final = response.choices[0].message.content
if not isinstance(text, str):
raise HTTPException(status_code=502, detail=f"shunt {label}: worker model returned no text")
return text
async def _read_upload_text(upload: UploadFile) -> str:
content: Final = await upload.read()
try:
return content.decode("utf-8")
except UnicodeDecodeError as e:
raise ProxyException(
message=f"'{upload.filename}' is not valid UTF-8 text",
type=ProxyErrorTypes.bad_request_error,
param="paths" if upload.filename else "file",
code=400,
) from e
@router.post(
"/v1/bulk_read",
dependencies=_AUTH_DEPENDENCIES,
tags=_SHUNT_TAGS,
)
@router.post(
"/bulk_read",
dependencies=_AUTH_DEPENDENCIES,
tags=_SHUNT_TAGS,
)
async def bulk_read(
request: Request,
router_name: Annotated[str, Query(alias="router")],
question: Annotated[str, Form()],
paths: Annotated[list[UploadFile], File()], # mutable-ok: FastAPI requires a list for a repeated file field
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
) -> str:
"""Summarize or answer a question about one or more files via a cheap worker model.
The paths shunt's generated command uploads are read here as plain UTF-8 text and never
written to disk; only the text and the question reach the worker model.
"""
llm_router, config = _worker_config(router_name, user_api_key_dict)
files: Final = MappingProxyType({upload.filename or "unnamed": await _read_upload_text(upload) for upload in paths})
message: Final = build_bulk_read_message(question=question, files=files)
verbose_proxy_logger.debug("shunt bulk_read: %s file(s) via %s", len(files), config.bulk_read_model)
return await _worker_text(
llm_router,
model=config.bulk_read_model,
system_prompt=BULK_READ_SYSTEM_PROMPT,
message=message,
team_id=user_api_key_dict.team_id,
label="bulk_read",
)
@router.post(
"/v1/code_write",
dependencies=_AUTH_DEPENDENCIES,
tags=_SHUNT_TAGS,
)
@router.post(
"/code_write",
dependencies=_AUTH_DEPENDENCIES,
tags=_SHUNT_TAGS,
)
async def code_write(
request: Request,
router_name: Annotated[str, Query(alias="router")],
spec: Annotated[str, Form()],
reference: Annotated[UploadFile, File()],
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
) -> str:
"""Generate boilerplate code matching a reference file's patterns, via a cheap worker model.
Returns the generated code as plain text. shunt's own script writes straight to disk since
it runs on the file's own machine; this has no local filesystem, so the client writes the
returned text itself (the `Bash` rewrite this backs redirects the curl output to `target`).
"""
llm_router, config = _worker_config(router_name, user_api_key_dict)
reference_content: Final = await _read_upload_text(reference)
message: Final = build_code_write_message(
spec=spec, reference_path=reference.filename or "", reference_content=reference_content
)
verbose_proxy_logger.debug("shunt code_write: reference=%s via %s", reference.filename, config.code_write_model)
return strip_code_fences(
await _worker_text(
llm_router,
model=config.code_write_model,
system_prompt=CODE_WRITE_SYSTEM_PROMPT,
message=message,
team_id=user_api_key_dict.team_id,
label="code_write",
)
)

49
litellm/proxy/shunt_endpoints/worker.py Normal file
View file

@ -0,0 +1,49 @@
"""
Prompt-building and output-cleaning for the shunt worker calls.
Ported verbatim from Spotify's shunt plugin README (see `auto_router_shunt.py`'s module
docstring for the source): the same system prompts, the same `<file path="...">` wrapping for
bulk-read, and the same fenced-code stripping for code-write.
"""
import re
from collections.abc import Mapping
from typing import Final
BULK_READ_SYSTEM_PROMPT: Final = (
"You are a precise code analyst. Read the provided files and answer the question "
"concisely. Output structured bullets only. No greetings, no prose, no preambles, no "
"summaries. Lead every bullet with the exact name, type, or line number. Use nested "
"bullets for details. Skip anything the caller did not ask for."
)
CODE_WRITE_SYSTEM_PROMPT: Final = (
"You generate code files based on a spec and reference files. Match the existing "
"patterns, conventions, naming, and style exactly. Output only the code, no explanations, "
"no markdown fences unless asked. If the spec is ambiguous, make reasonable choices that "
"match the patterns in the reference code."
)
WORKER_TEMPERATURE: Final = 0.2
_FENCE_LINE: Final = re.compile(r"^```.*$", re.MULTILINE)
def build_bulk_read_message(question: str, files: Mapping[str, str]) -> str:
"""The user message for a bulk-read call: each file wrapped in `<file path="...">` tags."""
file_blocks: Final = "".join(f'<file path="{path}">\n{content}\n</file>\n\n' for path, content in files.items())
return f"{file_blocks}Question: {question}"
def build_code_write_message(spec: str, reference_path: str, reference_content: str) -> str:
"""The user message for a code-write call: the spec plus one reference file."""
return f"Spec: {spec}\n\nReference:\n{reference_content}" if reference_path else f"Spec: {spec}"
def strip_code_fences(text: str) -> str:
"""Remove markdown code-fence lines, matching shunt's `sed '/^```/d'` post-processing.
The model is instructed not to wrap output in fences, but does anyway often enough that
shunt's own script strips them unconditionally rather than trusting the instruction.
"""
return _FENCE_LINE.sub("", text).strip()

5
litellm/types/router.py
View file

@ -364,6 +364,11 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams):
# request's own compression guardrails apply to both, as they always have.
auto_router_routing_compression: str | None = None
auto_router_model_compression: str | None = None
# Shunt: bounds large file/shell reads and delegates bulk-read/code-write tool calls to a
# cheap worker model. Presence of auto_router_shunt_min_lines arms it for this marker.
auto_router_shunt_min_lines: int | None = None
auto_router_shunt_bulk_read_model: str | None = None
auto_router_shunt_code_write_model: str | None = None
# complexity-router params
complexity_router_config: dict | None = None

3
litellm/types/utils.py
View file

@ -3789,6 +3789,9 @@ all_litellm_params = (
"auto_router_max_input_chars",
"auto_router_routing_compression",
"auto_router_model_compression",
"auto_router_shunt_min_lines",
"auto_router_shunt_bulk_read_model",
"auto_router_shunt_code_write_model",
"complexity_router_config",
"complexity_router_default_model",
"adaptive_router_config",

57
tests/test_litellm/proxy/client/cli/autoroute/test_settings.py
View file

@ -1,5 +1,9 @@
import json
from litellm.proxy.client.cli.commands.autoroute.settings import (
ANTHROPIC_DEFAULT_MODEL_ENV_KEYS,
SHUNT_BASH_ALLOW_RULES,
merge_claude_settings_shunt_permissions,
merge_claude_settings_static_token,
)
@ -61,3 +65,56 @@ def test_overrides_a_preexisting_default_model_env_var():
settings = {"env": {"ANTHROPIC_DEFAULT_SONNET_MODEL": "claude-opus-4-8"}}
merged = merge_claude_settings_static_token(settings, "http://127.0.0.1:4000", "token-abc")
assert merged["env"]["ANTHROPIC_DEFAULT_SONNET_MODEL"] == "autorouter"
def test_adds_shunt_allow_rules_to_empty_settings():
merged = merge_claude_settings_shunt_permissions({})
assert tuple(merged["permissions"]["allow"]) == SHUNT_BASH_ALLOW_RULES
def test_preserves_unrelated_top_level_keys_for_permissions_merge():
merged = merge_claude_settings_shunt_permissions({"theme": "dark"})
assert merged["theme"] == "dark"
def test_preserves_existing_allow_rules():
settings = {"permissions": {"allow": ["Bash(npm run *)"]}}
merged = merge_claude_settings_shunt_permissions(settings)
assert "Bash(npm run *)" in merged["permissions"]["allow"]
for rule in SHUNT_BASH_ALLOW_RULES:
assert rule in merged["permissions"]["allow"]
def test_preserves_deny_and_ask_rules():
# Regression: a naive {**settings, "permissions": {...}} replaces the whole permissions
# object, silently dropping deny/ask rules the caller already had.
settings = {"permissions": {"deny": ["Bash(git push *)"], "ask": ["Bash(rm *)"]}}
merged = merge_claude_settings_shunt_permissions(settings)
assert merged["permissions"]["deny"] == ["Bash(git push *)"]
assert merged["permissions"]["ask"] == ["Bash(rm *)"]
for rule in SHUNT_BASH_ALLOW_RULES:
assert rule in merged["permissions"]["allow"]
def test_does_not_duplicate_shunt_rules_already_present():
settings = {"permissions": {"allow": list(SHUNT_BASH_ALLOW_RULES)}}
merged = merge_claude_settings_shunt_permissions(settings)
assert tuple(merged["permissions"]["allow"]) == SHUNT_BASH_ALLOW_RULES
def test_merged_settings_survive_the_json_round_trip_commands_py_writes():
# Regression: the merge returns MappingProxyType nested at every level, which the JSON
# encoder rejects without `default=dict` -- the exact call commands.py makes. Without it
# `lite autoroute up` raises instead of writing settings.json.
settings = {"permissions": {"allow": ["Bash(npm run *)"], "deny": ["Bash(rm *)"]}, "theme": "dark"}
merged = merge_claude_settings_shunt_permissions(settings)
reloaded = json.loads(json.dumps(merged, default=dict))
assert reloaded["theme"] == "dark"
assert reloaded["permissions"]["deny"] == ["Bash(rm *)"]
assert reloaded["permissions"]["allow"] == ["Bash(npm run *)", *SHUNT_BASH_ALLOW_RULES]
def test_does_not_mutate_input_for_permissions_merge():
settings = {"permissions": {"allow": ["Bash(npm run *)"], "deny": ["Bash(rm *)"]}}
merge_claude_settings_shunt_permissions(settings)
assert settings == {"permissions": {"allow": ["Bash(npm run *)"], "deny": ["Bash(rm *)"]}}

355
tests/test_litellm/proxy/guardrails/test_auto_router_shunt.py Normal file
View file

@ -0,0 +1,355 @@
"""Unit tests for litellm.proxy.guardrails.auto_router_shunt."""
import json
from typing import Any
import pytest
from litellm.proxy.guardrails.auto_router_shunt import (
BULK_READ_TOOL_NAME,
CODE_WRITE_TOOL_NAME,
ShuntConfig,
ShuntGuardrail,
shunt_config_for_model,
)
from litellm.types.utils import ChatCompletionMessageToolCall, Choices, Function, Message, ModelResponse
class _FakeRouter:
"""Minimal stand-in for litellm.Router.get_model_list, mirroring
test_auto_router_compression.py's _FakeRouter."""
def __init__(self, deployments: list[dict[str, Any]]):
self._deployments = deployments
def get_model_list(self, model_name, team_id=None):
return [d for d in self._deployments if d.get("model_name") == model_name]
def _marker(shunt_fields: dict[str, Any], tags: list[str] | None = None) -> dict[str, Any]:
return {
"model_name": "shunt",
"litellm_params": {
"model": "auto_router/complexity_router",
"complexity_router_default_model": "claude-haiku-4-5",
**shunt_fields,
**({"tags": tags} if tags is not None else {}),
},
}
class TestShuntConfigForModel:
def test_no_router_returns_none(self):
assert shunt_config_for_model(llm_router=None, model_alias="shunt", team_id=None, request_tags=()) is None
def test_no_marker_deployment_returns_none(self):
router = _FakeRouter([{"model_name": "shunt", "litellm_params": {"model": "openai/gpt-4o-mini"}}])
assert shunt_config_for_model(llm_router=router, model_alias="shunt", team_id=None, request_tags=()) is None
def test_marker_without_min_lines_is_unarmed(self):
router = _FakeRouter([_marker({})])
assert shunt_config_for_model(llm_router=router, model_alias="shunt", team_id=None, request_tags=()) is None
def test_armed_marker_falls_back_to_default_model_for_worker_models(self):
router = _FakeRouter([_marker({"auto_router_shunt_min_lines": 350})])
config = shunt_config_for_model(llm_router=router, model_alias="shunt", team_id=None, request_tags=())
assert config == ShuntConfig(
min_lines=350, bulk_read_model="claude-haiku-4-5", code_write_model="claude-haiku-4-5"
)
def test_explicit_worker_models_override_the_default(self):
router = _FakeRouter(
[
_marker(
{
"auto_router_shunt_min_lines": 200,
"auto_router_shunt_bulk_read_model": "gpt-5.6-luna",
}
)
]
)
config = shunt_config_for_model(llm_router=router, model_alias="shunt", team_id=None, request_tags=())
assert config.bulk_read_model == "gpt-5.6-luna"
assert config.code_write_model == "claude-haiku-4-5"
def test_picks_the_marker_whose_tags_the_request_carries(self):
router = _FakeRouter(
[
_marker({"auto_router_shunt_min_lines": 100}, tags=["eu"]),
_marker({"auto_router_shunt_min_lines": 999}, tags=["us"]),
]
)
eu = shunt_config_for_model(llm_router=router, model_alias="shunt", team_id=None, request_tags=("eu",))
assert eu.min_lines == 100
class TestAsyncPreCallHook:
@pytest.mark.asyncio
async def test_unarmed_returns_none(self):
guardrail = ShuntGuardrail()
data = {"model": "some-model", "messages": []}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="acompletion"
)
assert result is None
@pytest.mark.asyncio
async def test_wrong_call_type_returns_none_without_touching_data(self, monkeypatch):
guardrail = ShuntGuardrail()
data = {"model": "shunt", "tools": []}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="amoderation"
)
assert result is None
assert data["tools"] == []
@pytest.mark.asyncio
async def test_armed_injects_anthropic_shape_tools(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
router = _FakeRouter([_marker({"auto_router_shunt_min_lines": 350})])
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", router)
guardrail = mod.ShuntGuardrail()
data = {"model": "shunt", "messages": []}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="anthropic_messages"
)
assert result is not None
tool_names = {tool["name"] for tool in result["tools"]}
assert tool_names == {BULK_READ_TOOL_NAME, CODE_WRITE_TOOL_NAME}
@pytest.mark.asyncio
async def test_armed_injects_openai_shape_tools_for_completion(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
router = _FakeRouter([_marker({"auto_router_shunt_min_lines": 350})])
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", router)
guardrail = mod.ShuntGuardrail()
data = {"model": "shunt", "messages": []}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="acompletion"
)
assert result is not None
function_names = {tool["function"]["name"] for tool in result["tools"]}
assert function_names == {BULK_READ_TOOL_NAME, CODE_WRITE_TOOL_NAME}
@pytest.mark.asyncio
async def test_appends_to_existing_tools_rather_than_replacing(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
router = _FakeRouter([_marker({"auto_router_shunt_min_lines": 350})])
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", router)
guardrail = mod.ShuntGuardrail()
existing_tool = {"name": "some_other_tool", "description": "x", "input_schema": {}}
data = {"model": "shunt", "tools": [existing_tool]}
result = await guardrail.async_pre_call_hook(
user_api_key_dict=None, cache=None, data=data, call_type="anthropic_messages"
)
tool_names = {tool["name"] for tool in result["tools"]}
assert "some_other_tool" in tool_names
assert BULK_READ_TOOL_NAME in tool_names
class TestAsyncPostCallSuccessHookAnthropicShape:
def _config(self) -> ShuntConfig:
return ShuntConfig(min_lines=350, bulk_read_model="claude-haiku-4-5", code_write_model="claude-haiku-4-5")
def _armed_request_data(self, model="shunt") -> dict:
return {
"model": model,
"proxy_server_request": {"url": "http://localhost:4000/v1/messages"},
"secret_fields": {"raw_headers": {"authorization": "Bearer sk-1234"}},
}
@pytest.mark.asyncio
async def test_unarmed_returns_response_untouched(self):
guardrail = ShuntGuardrail()
response = {"content": [{"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "x.py"}}]}
result = await guardrail.async_post_call_success_hook(
data={"model": "some-model"}, user_api_key_dict=None, response=response
)
assert result["content"][0]["name"] == "Read"
@pytest.mark.asyncio
async def test_rewrites_untargeted_read_to_bash(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {
"content": [{"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "litellm/router.py"}}]
}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
block = result["content"][0]
assert block["name"] == "Bash"
assert "curl" in block["input"]["command"]
assert "litellm/router.py" in block["input"]["command"]
@pytest.mark.asyncio
async def test_leaves_targeted_read_untouched(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {
"content": [
{"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "x.py", "offset": 10}}
]
}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
assert result["content"][0]["name"] == "Read"
@pytest.mark.asyncio
async def test_leaves_endpoints_unresolvable_response_untouched(self, monkeypatch):
"""No base_url/auth_header recoverable -> the rewrite is skipped rather than shipped broken."""
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {
"content": [{"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "litellm/router.py"}}]
}
result = await guardrail.async_post_call_success_hook(
data={"model": "shunt"}, user_api_key_dict=None, response=response
)
assert result["content"][0]["name"] == "Read"
@pytest.mark.asyncio
async def test_rewrites_explicit_bulk_read_call(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {
"content": [
{
"type": "tool_use",
"id": "t1",
"name": BULK_READ_TOOL_NAME,
"input": {"question": "what does this do", "paths": ["a.py", "b.py"]},
}
]
}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
block = result["content"][0]
assert block["name"] == "Bash"
assert "paths[]=@a.py" in block["input"]["command"]
assert "paths[]=@b.py" in block["input"]["command"]
@pytest.mark.asyncio
async def test_rewrites_explicit_code_write_call_with_target(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {
"content": [
{
"type": "tool_use",
"id": "t1",
"name": CODE_WRITE_TOOL_NAME,
"input": {"spec": "write tests", "reference": "tests/y_test.py", "target": "tests/x_test.py"},
}
]
}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
block = result["content"][0]
assert block["name"] == "Bash"
assert '> "tests/x_test.py"' in block["input"]["command"]
@pytest.mark.asyncio
async def test_rewrites_bare_bash_read_command(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {
"content": [
{"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": "cat litellm/router.py"}}
]
}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
assert "wc -l" in result["content"][0]["input"]["command"]
@pytest.mark.asyncio
async def test_leaves_piped_bash_command_untouched(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
original_command = "cat litellm/router.py | grep foo"
response = {"content": [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": original_command}}]}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
assert result["content"][0]["input"]["command"] == original_command
@pytest.mark.asyncio
async def test_leaves_non_tool_use_block_untouched(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = {"content": [{"type": "text", "text": "hello"}]}
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
assert result["content"][0]["type"] == "text"
assert result["content"][0]["text"] == "hello"
class TestAsyncPostCallSuccessHookOpenAIShape:
def _config(self) -> ShuntConfig:
return ShuntConfig(min_lines=350, bulk_read_model="claude-haiku-4-5", code_write_model="claude-haiku-4-5")
def _armed_request_data(self) -> dict:
return {
"model": "shunt",
"proxy_server_request": {"url": "http://localhost:4000/v1/chat/completions"},
"secret_fields": {"raw_headers": {"authorization": "Bearer sk-1234"}},
}
@pytest.mark.asyncio
async def test_rewrites_untargeted_read_tool_call(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
tool_call = ChatCompletionMessageToolCall(
id="call_1", function=Function(name="Read", arguments=json.dumps({"file_path": "litellm/router.py"}))
)
response = ModelResponse(
choices=[Choices(index=0, message=Message(role="assistant", tool_calls=[tool_call]))]
)
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
rewritten = result.choices[0].message.tool_calls[0]
assert rewritten.function.name == "Bash"
parsed = json.loads(rewritten.function.arguments)
assert "curl" in parsed["command"]
@pytest.mark.asyncio
async def test_leaves_response_with_no_tool_calls_untouched(self, monkeypatch):
import litellm.proxy.guardrails.auto_router_shunt as mod
monkeypatch.setattr(mod, "_resolve_shunt_config", lambda data: self._config())
guardrail = mod.ShuntGuardrail()
response = ModelResponse(choices=[Choices(index=0, message=Message(role="assistant", content="hi"))])
result = await guardrail.async_post_call_success_hook(
data=self._armed_request_data(), user_api_key_dict=None, response=response
)
assert result.choices[0].message.content == "hi"

173
tests/test_litellm/proxy/guardrails/test_shunt_rewrite.py Normal file
View file

@ -0,0 +1,173 @@
"""Unit tests for litellm.proxy.guardrails.shunt_rewrite."""
import subprocess
import pytest
from litellm.proxy.guardrails.shunt_rewrite import (
ShuntBashRewrite,
build_bounded_read_command,
build_bulk_read_command,
build_code_write_command,
extract_bare_read_path,
is_targeted_read,
)
class TestIsTargetedRead:
def test_neither_set_is_untargeted(self):
assert is_targeted_read(None, None) is False
def test_offset_set_is_targeted(self):
assert is_targeted_read(100, None) is True
def test_limit_set_is_targeted(self):
assert is_targeted_read(None, 50) is True
def test_both_set_is_targeted(self):
assert is_targeted_read(100, 50) is True
def test_offset_zero_is_targeted(self):
"""Matches shunt's own documented bypass: offset:0 counts as targeted."""
assert is_targeted_read(0, None) is True
def test_limit_zero_is_targeted(self):
assert is_targeted_read(None, 0) is True
class TestExtractBareReadPath:
# Ported one-for-one from shunt's own evals/bash-hook-evals.json fixture cases.
@pytest.mark.parametrize(
"command,expected",
[
("cat file.txt", "file.txt"),
("cat -n file.txt", "file.txt"),
("head file.txt", "file.txt"),
("head -100 file.txt", "file.txt"),
("tail file.txt", "file.txt"),
("less file.txt", "file.txt"),
("more file.txt", "file.txt"),
('cat "file.txt"', "file.txt"),
("cat file.txt | grep export", None),
("cat file.txt > /tmp/out.txt", None),
("git status", None),
("grep -n 'export' file.txt", None),
("cat /tmp/does-not-exist.txt", "/tmp/does-not-exist.txt"),
("", None),
],
)
def test_ported_shunt_eval_cases(self, command: str, expected: str | None):
assert extract_bare_read_path(command) == expected
def test_fixes_shunts_documented_parser_bug(self):
"""shunt's own parser returns "5" for `head -n 5 file` (a flag's value token mistaken
for the path); this port skips a bare-numeric token instead."""
assert extract_bare_read_path("head -n 5 file.txt") == "file.txt"
def test_fixes_tails_plus_n_follow_syntax_the_same_way(self):
assert extract_bare_read_path("tail -n +5 file.log") == "file.log"
def test_no_non_flag_argument_returns_none(self):
assert extract_bare_read_path("cat -n") is None
def _bounded_read(
path: str = "litellm/router.py",
question: str = "Summarize this file's structure.",
min_lines: int = 350,
bulk_read_endpoint: str = "http://localhost:4000/v1/bulk_read",
auth_header: str = "Bearer sk-1234",
) -> ShuntBashRewrite:
return build_bounded_read_command(
path=path,
question=question,
min_lines=min_lines,
bulk_read_endpoint=bulk_read_endpoint,
auth_header=auth_header,
)
def _assert_valid_bash(command: str) -> None:
result = subprocess.run(["bash", "-n", "-c", command], capture_output=True, text=True, check=False)
assert result.returncode == 0, result.stderr
class TestBuildBoundedReadCommand:
def test_returns_a_shunt_bash_rewrite(self):
assert isinstance(_bounded_read(), ShuntBashRewrite)
def test_command_is_valid_bash(self):
_assert_valid_bash(_bounded_read().command)
def test_command_names_the_threshold(self):
assert "-gt 350" in _bounded_read(min_lines=350).command
def test_command_names_the_endpoint(self):
endpoint = "http://localhost:4000/v1/bulk_read?router=shunt"
assert endpoint in _bounded_read(bulk_read_endpoint=endpoint).command
def test_command_falls_back_to_cat_for_small_files(self):
assert 'else cat "litellm/router.py"; fi' in _bounded_read().command
def test_command_escapes_embedded_double_quotes_in_path(self):
_assert_valid_bash(_bounded_read(path='weird"file.py').command)
def test_note_mentions_the_threshold(self):
assert "200" in _bounded_read(min_lines=200).note
class TestBuildBulkReadCommand:
def test_unconditional_no_size_check(self):
rewrite = build_bulk_read_command(
question="what does this do",
paths=["a.py", "b.py"],
bulk_read_endpoint="http://localhost:4000/v1/bulk_read",
auth_header="Bearer sk-1234",
)
assert "wc -l" not in rewrite.command
assert "if [" not in rewrite.command
def test_includes_every_path(self):
rewrite = build_bulk_read_command(
question="q",
paths=["a.py", "b.py", "c.py"],
bulk_read_endpoint="http://localhost:4000/v1/bulk_read",
auth_header="Bearer sk-1234",
)
for path in ("a.py", "b.py", "c.py"):
assert f"paths[]=@{path}" in rewrite.command
def test_command_is_valid_bash(self):
rewrite = build_bulk_read_command(
question="q", paths=["a.py"], bulk_read_endpoint="http://localhost:4000/v1/bulk_read", auth_header="x"
)
_assert_valid_bash(rewrite.command)
class TestBuildCodeWriteCommand:
def test_no_target_outputs_to_stdout(self):
rewrite = build_code_write_command(
spec="write tests",
reference="tests/y_test.py",
target=None,
code_write_endpoint="http://localhost:4000/v1/code_write",
auth_header="Bearer sk-1234",
)
assert ">" not in rewrite.command
def test_target_redirects_curl_output_to_the_file(self):
rewrite = build_code_write_command(
spec="write tests",
reference="tests/y_test.py",
target="tests/x_test.py",
code_write_endpoint="http://localhost:4000/v1/code_write",
auth_header="Bearer sk-1234",
)
assert '> "tests/x_test.py"' in rewrite.command
@pytest.mark.parametrize("target", [None, "tests/x_test.py"])
def test_command_is_valid_bash_with_and_without_target(self, target: str | None):
rewrite = build_code_write_command(
spec="s", reference="r.py", target=target, code_write_endpoint="http://x/v1/code_write", auth_header="x"
)
_assert_valid_bash(rewrite.command)

53
ui/litellm-dashboard/src/components/add_model/ComplexityRouterConfig.tsx
View file

@ -52,6 +52,9 @@ import SemanticKeywordMatching from "./SemanticKeywordMatching";
import { type DimensionWeights, type TierBoundaries, type TokenThresholds } from "./heuristic_scoring_knobs";
import CompressionControls from "./CompressionControls";
import { type AutoRouterCompressionState, DEFAULT_AUTO_ROUTER_COMPRESSION } from "./buildAutoRouterCompression";
import ShuntControls from "./ShuntControls";
import { type AutoRouterShuntState, DEFAULT_AUTO_ROUTER_SHUNT } from "./buildAutoRouterShunt";
import { TIER_DESCRIPTIONS, TIER_KEYS, effectiveTierLabel } from "./tier_descriptions";
export type { DimensionWeights, TierBoundaries, TokenThresholds };
export type { CustomTierSet, TierRow } from "./tier_rows";
@ -487,39 +490,16 @@ interface ComplexityRouterConfigProps {
// pre-routing strategy, not just the complexity router.
autoRouterCompression?: AutoRouterCompressionState;
onAutoRouterCompressionChange?: (state: AutoRouterCompressionState) => void;
// Optional: same reasoning as autoRouterCompression - shunt arms via flat litellm_params
// fields shared by every pre-routing strategy, not a complexity_router_config setting.
autoRouterShunt?: AutoRouterShuntState;
onAutoRouterShuntChange?: (state: AutoRouterShuntState) => void;
showValidationErrors?: boolean;
}
export const TIER_DESCRIPTIONS: Record<
keyof ComplexityTiers,
{ label: string; description: string; examples: string }
> = {
SIMPLE: {
label: "Simple",
description: "Basic questions, greetings, simple factual queries",
examples: '"Hello!", "What is Python?", "Thanks!"',
},
MEDIUM: {
label: "Medium",
description: "Standard queries requiring some reasoning or explanation",
examples: '"Explain how REST APIs work", "Debug this error"',
},
COMPLEX: {
label: "Complex",
description: "Technical, multi-part requests requiring deep knowledge",
examples: '"Design a microservices architecture", "Implement a rate limiter"',
},
REASONING: {
label: "Reasoning",
description: "Chain-of-thought, analysis, explicit reasoning requests",
examples: '"Think step by step...", "Analyze the pros and cons..."',
},
};
export const TIER_KEYS = Object.keys(TIER_DESCRIPTIONS) as Array<keyof ComplexityTiers>;
export const effectiveTierLabel = (tier: keyof ComplexityTiers, tierLabels: ComplexityTierLabels | undefined): string =>
tierLabels?.[tier]?.trim() || TIER_DESCRIPTIONS[tier].label;
// Re-exported so the many existing importers of these from this module keep working; the data
// itself lives in ./tier_descriptions to keep this component under the max-lines gate.
export { TIER_DESCRIPTIONS, TIER_KEYS, effectiveTierLabel };
export const DEFAULT_HEURISTIC_FIRST_MAX_TIER = "SIMPLE";
@ -591,6 +571,8 @@ const ComplexityRouterConfig: React.FC<ComplexityRouterConfigProps> = ({
onEscalationKeywordsChange,
autoRouterCompression = DEFAULT_AUTO_ROUTER_COMPRESSION,
onAutoRouterCompressionChange,
autoRouterShunt = DEFAULT_AUTO_ROUTER_SHUNT,
onAutoRouterShuntChange,
showValidationErrors = false,
}) => {
const customTierSet = value.custom_tier_set;
@ -875,6 +857,17 @@ const ComplexityRouterConfig: React.FC<ComplexityRouterConfigProps> = ({
},
]
: []),
...(onAutoRouterShuntChange
? [
{
key: "shunt",
label: <strong className="text-foreground font-semibold">Advanced: Shunt</strong>,
children: (
<ShuntControls value={autoRouterShunt} onChange={onAutoRouterShuntChange} modelInfo={modelInfo} />
),
},
]
: []),
...(onKeywordTierRulesChange || onSemanticMatchingEnabledChange
? [
{

78
ui/litellm-dashboard/src/components/add_model/ShuntControls.tsx Normal file
View file

@ -0,0 +1,78 @@
import { Info } from "lucide-react";
import { SimpleTooltip } from "@/components/ui/tooltip";
import { SearchSelect } from "@/components/shared/SearchSelect";
import { Input } from "@/components/ui/input";
import React from "react";
import { ModelGroup } from "@/components/llm_calls/fetch_models";
import { AutoRouterShuntState, DEFAULT_AUTO_ROUTER_SHUNT_MIN_LINES } from "./buildAutoRouterShunt";
interface ShuntControlsProps {
value: AutoRouterShuntState;
onChange: (state: AutoRouterShuntState) => void;
modelInfo: ModelGroup[];
}
const ShuntControls: React.FC<ShuntControlsProps> = ({ value, onChange, modelInfo }) => {
const { minLines, bulkReadModel, codeWriteModel } = value;
const armed = minLines !== undefined;
const chatModelOptions = Array.from(
new Set(
modelInfo.filter((model) => model.mode === undefined || model.mode === "chat").map((model) => model.model_group),
),
).map((model_group) => ({ value: model_group, label: model_group }));
return (
<div className="space-y-4">
<div className="flex items-center gap-2">
<span className="text-sm font-medium">Bounded-read threshold</span>
<SimpleTooltip content="Files this router's own Read/Bash tool calls read above this many lines are delegated to a cheaper model instead of entering the routed model's context.">
<Info className="size-4 text-muted-foreground" />
</SimpleTooltip>
</div>
<Input
type="number"
value={minLines ?? ""}
onChange={(event) =>
onChange({
...value,
minLines: event.target.value === "" ? undefined : event.target.valueAsNumber,
})
}
placeholder={String(DEFAULT_AUTO_ROUTER_SHUNT_MIN_LINES)}
min={1}
aria-label="Bounded-read threshold"
className="w-full"
/>
{armed && (
<div className="grid gap-4 md:grid-cols-2 mt-4 pt-4 border-t border-border">
<div>
<span className="mb-1 block text-sm font-medium">Bulk-read model</span>
<SearchSelect
options={chatModelOptions}
value={bulkReadModel ?? ""}
onValueChange={(model) => onChange({ ...value, bulkReadModel: model === "" ? undefined : model })}
placeholder="Use this router's default model"
emptyText="No chat models found"
aria-label="Bulk-read model"
/>
</div>
<div>
<span className="mb-1 block text-sm font-medium">Code-write model</span>
<SearchSelect
options={chatModelOptions}
value={codeWriteModel ?? ""}
onValueChange={(model) => onChange({ ...value, codeWriteModel: model === "" ? undefined : model })}
placeholder="Use this router's default model"
emptyText="No chat models found"
aria-label="Code-write model"
/>
</div>
</div>
)}
</div>
);
};
export default ShuntControls;

104
ui/litellm-dashboard/src/components/add_model/add_auto_router_tab.test.tsx
View file

@ -771,6 +771,7 @@ describe("AddAutoRouterTab", () => {
"Gemini Family",
"Lite",
"OpenAI Family",
"Shunt",
"Custom Configuration",
]);
});
@ -1027,6 +1028,105 @@ describe("AddAutoRouterTab", () => {
});
});
// Shunt is a partial-fit preset (see isPartialFitPreset in @/lib/autorouter_presets): unlike
// the family presets above, it stays selectable even when none of its own named tier models
// are available, as long as the caller has some chat model at all.
describe("shunt preset", () => {
it("stays selectable with only unrelated chat models, unlike an ordinary family preset", async () => {
mockFetchAvailableModels.mockResolvedValue([{ model_group: "unknown-model-a", mode: "chat" }]);
renderWithProviders(<Harness />);
openTemplateDropdown();
await waitFor(() => expect(optionByLabel("Anthropic Family")).toHaveTextContent("Missing:"));
await waitFor(() => expect(isOptionDisabled(optionByLabel("Shunt")!)).toBe(false));
});
it("greys out only when the caller has no chat model at all", async () => {
mockFetchAvailableModels.mockResolvedValue([]);
renderWithProviders(<Harness />);
openTemplateDropdown();
await waitFor(() => expect(isOptionDisabled(optionByLabel("Shunt")!)).toBe(true));
});
it("expands detailed configuration with empty tiers when none of its own models resolve", async () => {
const user = userEvent.setup();
const actual = await vi.importActual<typeof import("./build_complexity_router_config")>(
"./build_complexity_router_config",
);
vi.mocked(getMissingTiersError).mockImplementation(actual.getMissingTiersError);
mockFetchAvailableModels.mockResolvedValue([{ model_group: "unknown-model-a", mode: "chat" }]);
renderWithProviders(<Harness />);
await waitForPresetEnabled("Shunt");
await selectTemplate("Shunt");
// Detailed Configuration auto-expands (none of Shunt's own tier models resolved, the same
// "needs your input" signal a deployment-matched family preset also expands for), and the
// submit stays blocked until a caller fills in a model for every tier - the same gate
// Custom Configuration is held to, since an unfilled tier is unfilled either way.
expect(screen.getByText("Complexity Tier Configuration")).toBeInTheDocument();
await user.type(screen.getByPlaceholderText(/smart_router/i), "shunt-empty-tiers");
expect(screen.getByRole("button", { name: /add auto router/i })).toBeDisabled();
});
it("carries the bounded-read threshold through to the create payload", async () => {
const user = userEvent.setup();
vi.mocked(getMissingTiersError).mockReturnValue(null);
mockFetchAvailableModels.mockResolvedValue([{ model_group: "some-chat-model", mode: "chat" }]);
renderWithProviders(<Harness />);
await waitForPresetEnabled("Shunt");
await selectTemplate("Shunt");
await user.type(screen.getByPlaceholderText(/smart_router/i), "shunt-router");
await user.click(screen.getByRole("button", { name: /add auto router/i }));
await waitFor(() => expect(handleAddAutoRouterSubmit).toHaveBeenCalled());
const submitted = vi.mocked(handleAddAutoRouterSubmit).mock.calls.at(-1)?.[0];
expect(submitted?.auto_router_shunt_min_lines).toBe(350);
});
it("omits shunt fields entirely for a router built from Custom Configuration", async () => {
const user = userEvent.setup();
vi.mocked(getMissingTiersError).mockReturnValue(null);
mockFetchAvailableModels.mockResolvedValue(ALL_FAMILY_MODELS);
renderWithProviders(<Harness />);
await waitForPresetEnabled("Anthropic Family");
await selectTemplate("Anthropic Family");
await user.type(screen.getByPlaceholderText(/smart_router/i), "anthropic-not-shunt");
await user.click(screen.getByRole("button", { name: /add auto router/i }));
await waitFor(() => expect(handleAddAutoRouterSubmit).toHaveBeenCalled());
const submitted = vi.mocked(handleAddAutoRouterSubmit).mock.calls.at(-1)?.[0];
expect(submitted?.auto_router_shunt_min_lines).toBeUndefined();
});
it("carries a chosen worker model through to the create payload", async () => {
const user = userEvent.setup();
vi.mocked(getMissingTiersError).mockReturnValue(null);
mockFetchAvailableModels.mockResolvedValue([{ model_group: "some-chat-model", mode: "chat" }]);
renderWithProviders(<Harness />);
await waitForPresetEnabled("Shunt");
await selectTemplate("Shunt");
await user.type(screen.getByPlaceholderText(/smart_router/i), "shunt-with-worker-model");
await user.click(screen.getByText("Advanced: Shunt"));
await chooseSelectOption(user, screen.getByRole("combobox", { name: "Bulk-read model" }), "some-chat-model");
await user.click(screen.getByRole("button", { name: /add auto router/i }));
await waitFor(() => expect(handleAddAutoRouterSubmit).toHaveBeenCalled());
const submitted = vi.mocked(handleAddAutoRouterSubmit).mock.calls.at(-1)?.[0];
expect(submitted?.auto_router_shunt_bulk_read_model).toBe("some-chat-model");
});
});
describe("default model pin", () => {
const PINNED_MODEL = "pinned-default-model";
@ -1196,8 +1296,12 @@ describe("AddAutoRouterTab", () => {
expect(isOptionDisabled(optionByLabel("Anthropic Family")!)).toBe(false);
});
const labels = visibleOptions().map((option) => option.querySelector(".font-medium")?.textContent);
// Shunt is a partial-fit preset (see isPartialFitPreset): it stays available as long as SOME
// chat model is registered, regardless of whether it's one of Shunt's own named models, so it
// sorts alongside Anthropic Family here rather than with the family presets that stay disabled.
expect(labels).toEqual([
"Anthropic Family",
"Shunt",
"1M Context",
"Gemini Family",
"Lite",

37
ui/litellm-dashboard/src/components/add_model/add_auto_router_tab.tsx
View file

@ -37,6 +37,12 @@ import {
buildAutoRouterCompressionParams,
DEFAULT_AUTO_ROUTER_COMPRESSION,
} from "./buildAutoRouterCompression";
import {
type AutoRouterShuntState,
buildAutoRouterShuntParams,
DEFAULT_AUTO_ROUTER_SHUNT,
shuntStateFromPreset,
} from "./buildAutoRouterShunt";
import { DEFAULT_MATCH_THRESHOLD } from "./SemanticKeywordMatching";
import {
BuildComplexityRouterConfigParams,
@ -63,6 +69,8 @@ import {
buildPresetPrefill,
buildModelAvailability,
deploymentRefsFromModelInfo,
hasNoUsableModelsAtAll,
isPartialFitPreset,
ModelAvailability,
PresetPrefill,
AutoRouterPreset,
@ -202,6 +210,7 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
const [autoRouterCompression, setAutoRouterCompression] = useState<AutoRouterCompressionState>(
DEFAULT_AUTO_ROUTER_COMPRESSION,
);
const [autoRouterShunt, setAutoRouterShunt] = useState<AutoRouterShuntState>(DEFAULT_AUTO_ROUTER_SHUNT);
const [showValidationErrors, setShowValidationErrors] = useState<boolean>(false);
const [editingTiers, setEditingTiers] = useState(false);
const [isSubmitting, setIsSubmitting] = useState(false);
@ -290,10 +299,23 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
// have been applied: while loading we withhold selection rather than let a caller pick a preset
// whose models we cannot yet verify, and a failed fetch leaves every preset unverifiable. This
// makes the load-race (pick during loading, then discover a missing model) unrepresentable.
//
// A partial-fit preset (isPartialFitPreset) is gated differently: its interception settings
// apply regardless of tier models, so it greys out only when the caller has no usable chat
// model at all, never merely because none of its own named models match the caller's fleet.
// Selecting it then leaves whichever tiers didn't resolve empty for the caller to fill in
// (see handlePresetChange), rather than blocking selection the way every other preset does.
const presetAvailability = React.useCallback(
(preset: AutoRouterPreset): PresetAvailability => {
if (modelsLoading) return { kind: "loading" };
if (modelsUnverifiable) return { kind: "unverifiable" };
if (isPartialFitPreset(preset)) {
if (hasNoUsableModelsAtAll(availability)) return { kind: "missing_models", models: ["any chat model"] };
// Expand Detailed Configuration when at least one tier didn't resolve to any of the
// preset's own models, the same "needs your input" signal viaDeployments already means
// for a fully-resolving preset - here it can mean the caller has to fill a tier by hand.
return { kind: "available", viaDeployments: getMissingModelsInPreset(preset, availability).length > 0 };
}
const missing = getMissingModelsInPreset(preset, availability);
if (missing.length > 0) return { kind: "missing_models", models: missing };
return {
@ -320,7 +342,9 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
[sortedPresetOptions],
);
const applyPrefill = (prefill: PresetPrefill) => {
// Every preset switch resets shunt state too (default: untouched), so switching away from
// Shunt to another preset can never leave its threshold/worker-model fields armed and stale.
const applyPrefill = (prefill: PresetPrefill, shunt: AutoRouterShuntState = DEFAULT_AUTO_ROUTER_SHUNT) => {
setEditingTiers(false);
setComplexityRouterConfig(prefill.complexityRouterConfig);
setCustomTechnicalKeywords(prefill.customTechnicalKeywords);
@ -329,6 +353,7 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
setEmbeddingModel(prefill.embeddingModel);
setMatchThreshold(prefill.matchThreshold);
setEscalationKeywords(prefill.escalationKeywords);
setAutoRouterShunt(shunt);
};
const handleAutomaticSetup = () => {
@ -355,7 +380,12 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
if (presetState.kind !== "available") return;
setSelectedPreset(presetKey);
applyPrefill(buildPresetPrefill(preset.complexity_router_config, availability));
applyPrefill(
buildPresetPrefill(preset.complexity_router_config, availability, {
dropUnresolvedTierEntries: isPartialFitPreset(preset),
}),
shuntStateFromPreset(preset.auto_router_shunt_min_lines),
);
setDetailsExpanded(presetState.viaDeployments);
};
@ -474,6 +504,7 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
complexity_router_config: complexityRouterConfigPayload,
model_access_group: form.getValues("model_access_group"),
...buildAutoRouterCompressionParams(autoRouterCompression),
...buildAutoRouterShuntParams(autoRouterShunt),
};
await handleAddAutoRouterSubmit(submitValues, accessToken, () => form.reset(EMPTY_FORM_VALUES), handleOk);
@ -681,6 +712,8 @@ const AddAutoRouterTab: React.FC<AddAutoRouterTabProps> = ({
onEscalationKeywordsChange={setEscalationKeywords}
autoRouterCompression={autoRouterCompression}
onAutoRouterCompressionChange={setAutoRouterCompression}
autoRouterShunt={autoRouterShunt}
onAutoRouterShuntChange={setAutoRouterShunt}
showValidationErrors={showValidationErrors}
/>
</div>

84
ui/litellm-dashboard/src/components/add_model/buildAutoRouterShunt.test.ts Normal file
View file

@ -0,0 +1,84 @@
import {
buildAutoRouterShuntParams,
DEFAULT_AUTO_ROUTER_SHUNT,
hydrateAutoRouterShunt,
shuntStateFromPreset,
} from "./buildAutoRouterShunt";
describe("buildAutoRouterShuntParams", () => {
it("omits all three keys when the threshold was never configured", () => {
expect(buildAutoRouterShuntParams(DEFAULT_AUTO_ROUTER_SHUNT)).toEqual({});
});
it("sends only the threshold when neither worker model is chosen", () => {
const params = buildAutoRouterShuntParams({ minLines: 350, bulkReadModel: undefined, codeWriteModel: undefined });
expect(params).toEqual({ auto_router_shunt_min_lines: 350 });
});
it("sends both worker models when both are chosen", () => {
const params = buildAutoRouterShuntParams({
minLines: 200,
bulkReadModel: "claude-haiku-4-5",
codeWriteModel: "gpt-5.6-luna",
});
expect(params).toEqual({
auto_router_shunt_min_lines: 200,
auto_router_shunt_bulk_read_model: "claude-haiku-4-5",
auto_router_shunt_code_write_model: "gpt-5.6-luna",
});
});
it("omits a worker model key when only the other is chosen", () => {
const params = buildAutoRouterShuntParams({
minLines: 200,
bulkReadModel: "claude-haiku-4-5",
codeWriteModel: undefined,
});
expect(params).toEqual({ auto_router_shunt_min_lines: 200, auto_router_shunt_bulk_read_model: "claude-haiku-4-5" });
});
it("arms at a threshold of 0, since only undefined means untouched", () => {
const params = buildAutoRouterShuntParams({ minLines: 0, bulkReadModel: undefined, codeWriteModel: undefined });
expect(params).toEqual({ auto_router_shunt_min_lines: 0 });
});
});
describe("hydrateAutoRouterShunt", () => {
it("returns the default (untouched) state when the threshold key is absent", () => {
expect(hydrateAutoRouterShunt({})).toEqual(DEFAULT_AUTO_ROUTER_SHUNT);
});
it("hydrates a stored threshold with no worker models set", () => {
const state = hydrateAutoRouterShunt({ auto_router_shunt_min_lines: 350 });
expect(state).toEqual({ minLines: 350, bulkReadModel: undefined, codeWriteModel: undefined });
});
it("hydrates a stored threshold with both worker models set", () => {
const state = hydrateAutoRouterShunt({
auto_router_shunt_min_lines: 200,
auto_router_shunt_bulk_read_model: "claude-haiku-4-5",
auto_router_shunt_code_write_model: "gpt-5.6-luna",
});
expect(state).toEqual({ minLines: 200, bulkReadModel: "claude-haiku-4-5", codeWriteModel: "gpt-5.6-luna" });
});
it("round-trips through buildAutoRouterShuntParams", () => {
const original = {
auto_router_shunt_min_lines: 350,
auto_router_shunt_bulk_read_model: "claude-haiku-4-5",
auto_router_shunt_code_write_model: "gpt-5.6-luna",
};
const rebuilt = buildAutoRouterShuntParams(hydrateAutoRouterShunt(original));
expect(rebuilt).toEqual(original);
});
});
describe("shuntStateFromPreset", () => {
it("stays untouched when the preset carries no threshold", () => {
expect(shuntStateFromPreset(undefined)).toEqual(DEFAULT_AUTO_ROUTER_SHUNT);
});
it("arms with the preset's threshold and no worker-model overrides", () => {
expect(shuntStateFromPreset(350)).toEqual({ minLines: 350, bulkReadModel: undefined, codeWriteModel: undefined });
});
});

60
ui/litellm-dashboard/src/components/add_model/buildAutoRouterShunt.ts Normal file
View file

@ -0,0 +1,60 @@
/**
* Maps the auto router's shunt form state to the three flat litellm_params keys the backend
* reads (litellm.proxy.guardrails.auto_router_shunt), and back.
*
* `minLines` being undefined means the section was never touched: all three keys are omitted
* from the payload, and shunt stays unarmed for this router. Once a caller sets a threshold,
* shunt is armed; the two worker-model fields are optional even then, since the backend
* defaults each to the router's own default model when unset (auto_router_default_model /
* complexity_router_default_model).
*/
export interface AutoRouterShuntState {
minLines: number | undefined;
bulkReadModel: string | undefined;
codeWriteModel: string | undefined;
}
export interface AutoRouterShuntLitellmParams {
auto_router_shunt_min_lines?: number;
auto_router_shunt_bulk_read_model?: string;
auto_router_shunt_code_write_model?: string;
}
export const DEFAULT_AUTO_ROUTER_SHUNT_MIN_LINES = 350;
export const DEFAULT_AUTO_ROUTER_SHUNT: AutoRouterShuntState = {
minLines: undefined,
bulkReadModel: undefined,
codeWriteModel: undefined,
};
export const buildAutoRouterShuntParams = (state: AutoRouterShuntState): AutoRouterShuntLitellmParams => {
if (state.minLines === undefined) return {};
return {
auto_router_shunt_min_lines: state.minLines,
...(state.bulkReadModel && { auto_router_shunt_bulk_read_model: state.bulkReadModel }),
...(state.codeWriteModel && { auto_router_shunt_code_write_model: state.codeWriteModel }),
};
};
export const hydrateAutoRouterShunt = (litellmParams: {
auto_router_shunt_min_lines?: number | null;
auto_router_shunt_bulk_read_model?: string | null;
auto_router_shunt_code_write_model?: string | null;
}): AutoRouterShuntState => {
const storedMinLines = litellmParams.auto_router_shunt_min_lines ?? undefined;
if (storedMinLines === undefined) return DEFAULT_AUTO_ROUTER_SHUNT;
return {
minLines: storedMinLines,
bulkReadModel: litellmParams.auto_router_shunt_bulk_read_model ?? undefined,
codeWriteModel: litellmParams.auto_router_shunt_code_write_model ?? undefined,
};
};
/** From a partial-fit preset's own top-level auto_router_shunt_min_lines (see
* autorouter_presets.ts's AutoRouterPreset.auto_router_shunt_min_lines), or untouched. */
export const shuntStateFromPreset = (presetMinLines: number | undefined): AutoRouterShuntState =>
presetMinLines === undefined
? DEFAULT_AUTO_ROUTER_SHUNT
: { minLines: presetMinLines, bulkReadModel: undefined, codeWriteModel: undefined };

6
ui/litellm-dashboard/src/components/add_model/handle_add_auto_router_submit.tsx
View file

@ -2,8 +2,9 @@ import { modelCreateCall } from "../networking";
import { toast } from "@/lib/toast";
import type { ComplexityRouterConfigPayload } from "./build_complexity_router_config";
import type { AutoRouterCompressionLitellmParams } from "./buildAutoRouterCompression";
import type { AutoRouterShuntLitellmParams } from "./buildAutoRouterShunt";
export interface AddAutoRouterValues extends AutoRouterCompressionLitellmParams {
export interface AddAutoRouterValues extends AutoRouterCompressionLitellmParams, AutoRouterShuntLitellmParams {
auto_router_name: string;
auto_router_default_model: string | undefined;
model_type: "complexity_router";
@ -27,6 +28,9 @@ export const handleAddAutoRouterSubmit = async (
complexity_router_default_model: values.auto_router_default_model,
auto_router_routing_compression: values.auto_router_routing_compression,
auto_router_model_compression: values.auto_router_model_compression,
auto_router_shunt_min_lines: values.auto_router_shunt_min_lines,
auto_router_shunt_bulk_read_model: values.auto_router_shunt_bulk_read_model,
auto_router_shunt_code_write_model: values.auto_router_shunt_code_write_model,
},
model_info: {
...(values.team_id ? { team_id: values.team_id } : {}),

32
ui/litellm-dashboard/src/components/add_model/tier_descriptions.ts Normal file
View file

@ -0,0 +1,32 @@
import type { ComplexityTierLabels, ComplexityTiers } from "./ComplexityRouterConfig";
export const TIER_DESCRIPTIONS: Record<
keyof ComplexityTiers,
{ label: string; description: string; examples: string }
> = {
SIMPLE: {
label: "Simple",
description: "Basic questions, greetings, simple factual queries",
examples: '"Hello!", "What is Python?", "Thanks!"',
},
MEDIUM: {
label: "Medium",
description: "Standard queries requiring some reasoning or explanation",
examples: '"Explain how REST APIs work", "Debug this error"',
},
COMPLEX: {
label: "Complex",
description: "Technical, multi-part requests requiring deep knowledge",
examples: '"Design a microservices architecture", "Implement a rate limiter"',
},
REASONING: {
label: "Reasoning",
description: "Chain-of-thought, analysis, explicit reasoning requests",
examples: '"Think step by step...", "Analyze the pros and cons..."',
},
};
export const TIER_KEYS = Object.keys(TIER_DESCRIPTIONS) as Array<keyof ComplexityTiers>;
export const effectiveTierLabel = (tier: keyof ComplexityTiers, tierLabels: ComplexityTierLabels | undefined): string =>
tierLabels?.[tier]?.trim() || TIER_DESCRIPTIONS[tier].label;

71
ui/litellm-dashboard/src/lib/autorouter_presets.test.ts
View file

@ -13,6 +13,8 @@ import {
buildModelAvailability,
deploymentRefsFromModelInfo,
normalizeModelName,
isPartialFitPreset,
hasNoUsableModelsAtAll,
} from "./autorouter_presets";
import { DEFAULT_MATCH_THRESHOLD } from "@/components/add_model/SemanticKeywordMatching";
import { DEFAULT_ESCALATION_KEYWORDS } from "@/components/add_model/EscalationKeywords";
@ -33,6 +35,7 @@ describe("autorouter_presets", () => {
"Gemini Family",
"Lite",
"OpenAI Family",
"Shunt",
]);
// Every preset carries all four fields the UI relies on; a JSON typo dropping one fails here.
for (const p of presets) {
@ -822,4 +825,72 @@ describe("autorouter_presets", () => {
expect(prefill.complexityRouterConfig.tier_model_params).toBeUndefined();
});
});
describe("isPartialFitPreset", () => {
it("marks the shunt preset as partial-fit", () => {
expect(isPartialFitPreset(getPresetByKey("shunt")!)).toBe(true);
});
it("does not mark an ordinary family preset as partial-fit", () => {
expect(isPartialFitPreset(getPresetByKey("anthropic_family")!)).toBe(false);
});
});
describe("hasNoUsableModelsAtAll", () => {
it("is true when the caller has registered no chat models", () => {
expect(hasNoUsableModelsAtAll(groupsOnly([]))).toBe(true);
});
it("is false as soon as any model group is registered, whether or not the preset names it", () => {
expect(hasNoUsableModelsAtAll(groupsOnly(["some-unrelated-model"]))).toBe(false);
});
});
describe("buildPresetPrefill with dropUnresolvedTierEntries", () => {
const config = {
tiers: {
SIMPLE: ["model-a", "model-b"],
MEDIUM: ["model-c"],
COMPLEX: ["model-d"],
REASONING: ["model-e"],
},
classifier_type: "heuristic" as const,
classification_mode: "every_request" as const,
session_affinity: false,
deployment_affinity: true,
};
it("drops a tier entry that fails to resolve instead of keeping the preset's literal string", () => {
const prefill = buildPresetPrefill(config, groupsOnly(["model-a"]), { dropUnresolvedTierEntries: true });
expect(prefill.complexityRouterConfig.tiers).toEqual({
SIMPLE: ["model-a"],
MEDIUM: [],
COMPLEX: [],
REASONING: [],
});
});
it("keeps every resolved entry in a tier that partially resolves", () => {
const prefill = buildPresetPrefill(config, groupsOnly(["model-a", "model-b"]), {
dropUnresolvedTierEntries: true,
});
expect(prefill.complexityRouterConfig.tiers.SIMPLE).toEqual(["model-a", "model-b"]);
});
it("defaults to the literal-fallback behavior when the option is omitted", () => {
const prefill = buildPresetPrefill(config, groupsOnly([]));
expect(prefill.complexityRouterConfig.tiers.SIMPLE).toEqual(["model-a", "model-b"]);
});
it("still rewrites a resolved entry to the caller's registered spelling", () => {
const versioned = {
...config,
tiers: { ...config.tiers, SIMPLE: ["claude-sonnet-4.5"] },
};
const prefill = buildPresetPrefill(versioned, groupsOnly(["claude-sonnet-4-5"]), {
dropUnresolvedTierEntries: true,
});
expect(prefill.complexityRouterConfig.tiers.SIMPLE).toEqual(["claude-sonnet-4-5"]);
});
});
});

50
ui/litellm-dashboard/src/lib/autorouter_presets.ts
View file

@ -28,8 +28,22 @@ export interface AutoRouterPreset {
label: string;
description: string;
complexity_router_config: ComplexityRouterConfigPayload;
// Present only on shunt-shaped presets: arms the shunt guardrail's bounded-read threshold.
// Lives at this top level, not inside complexity_router_config, because it maps to litellm_params
// fields (auto_router_shunt_*) rather than complexity_router_config settings.
auto_router_shunt_min_lines?: number;
}
// Presets whose tier models are suggestions rather than a hard requirement: shunt's value is the
// I/O interception it arms regardless of which tier models resolve, so a preset carrying this key
// should never grey out purely because none of ITS named tier models match the caller's fleet -
// only when the caller has no usable chat model at all. Keyed by preset.key rather than a field on
// the catalog JSON: this is a dashboard policy decision about how strictly to gate selection, not
// something a preset publisher should be able to opt into by editing the catalog.
const PARTIAL_FIT_PRESET_KEYS: ReadonlySet<string> = new Set(["shunt"]);
export const isPartialFitPreset = (preset: AutoRouterPreset): boolean => PARTIAL_FIT_PRESET_KEYS.has(preset.key);
export type AutoRouterPresetsResponse = Record<string, Omit<AutoRouterPreset, "key">>;
export const hydratePresets = (raw: AutoRouterPresetsResponse): AutoRouterPreset[] =>
@ -174,6 +188,11 @@ export const getRequiredModelsInPreset = (preset: AutoRouterPreset): Set<string>
export const getMissingModelsInPreset = (preset: AutoRouterPreset, availability: ModelAvailability): string[] =>
getMissingModels(preset.complexity_router_config, availability);
// The gate a partial-fit preset (see isPartialFitPreset) is selectable under: the proxy has to
// have registered SOME chat model, but not specifically one this preset names - unlike
// getMissingModelsInPreset's all-or-nothing check, which is still what every other preset uses.
export const hasNoUsableModelsAtAll = (availability: ModelAvailability): boolean => availability.modelGroups.size === 0;
// Checks the config actually being built (whether it arrived via a preset prefill or was typed by
// hand - the two are indistinguishable once the caller has started editing), not a preset's
// original bundled model list. Only counts classifier_llm_config/embedding_model as referenced
@ -191,15 +210,13 @@ export const getReferencedModelsError = (
},
availability: ModelAvailability,
): string | null => {
const missing = getMissingModels(
{
tiers: params.tiers,
default_model: params.defaultModel,
classifier_llm_config: usesLlmClassifier(params.classifierType) ? params.classifierLlmConfig : undefined,
embedding_model: params.semanticMatchingEnabled ? params.embeddingModel : undefined,
},
availability,
);
const referenced = {
tiers: params.tiers,
default_model: params.defaultModel,
classifier_llm_config: usesLlmClassifier(params.classifierType) ? params.classifierLlmConfig : undefined,
embedding_model: params.semanticMatchingEnabled ? params.embeddingModel : undefined,
};
const missing = getMissingModels(referenced, availability);
return missing.length > 0 ? `Model(s) no longer available: ${missing.join(", ")}` : null;
};
@ -240,9 +257,22 @@ export const buildEmptyPrefill = (): PresetPrefill => ({
export const buildPresetPrefill = (
config: ComplexityRouterConfigPayload,
availability: ModelAvailability,
// Partial-fit presets (see isPartialFitPreset) drop a tier entry that fails to resolve
// instead of keeping the preset's own unresolved string, so a caller who owns none of the
// preset's named models for a tier gets an empty picker to fill in rather than a config that
// silently names a model they don't have. Off by default: every other preset call site
// already guarantees full resolution before calling this (presetAvailability), so keeping the
// literal fallback there is unreachable, not a behavior change.
{ dropUnresolvedTierEntries = false }: { dropUnresolvedTierEntries?: boolean } = {},
): PresetPrefill => {
const resolve = (model: string): string => resolveAvailableModel(model, availability) ?? model;
const resolveTier = (models: string[]): string[] => models.map(resolve);
const resolveTier = (models: string[]): string[] =>
dropUnresolvedTierEntries
? models.flatMap((model) => {
const resolved = resolveAvailableModel(model, availability);
return resolved ? [resolved] : [];
})
: models.map(resolve);
// Params key on the model name the preset spells while every tier entry is rewritten to the
// caller's registered spelling, so the keys have to be rewritten the same way. Otherwise
// serializeTierModelConfigs drops them for naming a model the tier no longer holds.

274
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -1819,6 +1819,29 @@ export interface paths {
patch?: never;
trace?: never;
};
"/bulk_read": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Bulk Read
* @description Summarize or answer a question about one or more files via a cheap worker model.
*
* The paths shunt's generated command uploads are read here as plain UTF-8 text and never
* written to disk; only the text and the question reach the worker model.
*/
post: operations["bulk_read_bulk_read_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/cache/delete": {
parameters: {
query?: never;
@ -2459,6 +2482,30 @@ export interface paths {
patch?: never;
trace?: never;
};
"/code_write": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Code Write
* @description Generate boilerplate code matching a reference file's patterns, via a cheap worker model.
*
* Returns the generated code as plain text. shunt's own script writes straight to disk since
* it runs on the file's own machine; this has no local filesystem, so the client writes the
* returned text itself (the `Bash` rewrite this backs redirects the curl output to `target`).
*/
post: operations["code_write_code_write_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/cohere/{endpoint}": {
parameters: {
query?: never;
@ -17527,6 +17574,29 @@ export interface paths {
patch?: never;
trace?: never;
};
"/v1/bulk_read": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Bulk Read
* @description Summarize or answer a question about one or more files via a cheap worker model.
*
* The paths shunt's generated command uploads are read here as plain UTF-8 text and never
* written to disk; only the text and the question reach the worker model.
*/
post: operations["bulk_read_v1_bulk_read_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/chat/completions": {
parameters: {
query?: never;
@ -17562,6 +17632,30 @@ export interface paths {
patch?: never;
trace?: never;
};
"/v1/code_write": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Code Write
* @description Generate boilerplate code matching a reference file's patterns, via a cheap worker model.
*
* Returns the generated code as plain text. shunt's own script writes straight to disk since
* it runs on the file's own machine; this has no local filesystem, so the client writes the
* returned text itself (the `Bash` rewrite this backs redirects the curl output to `target`).
*/
post: operations["code_write_v1_code_write_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/completions": {
parameters: {
query?: never;
@ -24013,6 +24107,34 @@ export interface components {
/** Team Id */
team_id?: string | null;
};
/** Body_bulk_read_bulk_read_post */
Body_bulk_read_bulk_read_post: {
/** Paths */
paths: string[];
/** Question */
question: string;
};
/** Body_bulk_read_v1_bulk_read_post */
Body_bulk_read_v1_bulk_read_post: {
/** Paths */
paths: string[];
/** Question */
question: string;
};
/** Body_code_write_code_write_post */
Body_code_write_code_write_post: {
/** Reference */
reference: string;
/** Spec */
spec: string;
};
/** Body_code_write_v1_code_write_post */
Body_code_write_v1_code_write_post: {
/** Reference */
reference: string;
/** Spec */
spec: string;
};
/** Body_convert_prompt_file_to_json_utils_dotprompt_json_converter_post */
Body_convert_prompt_file_to_json_utils_dotprompt_json_converter_post: {
/** File */
@ -29306,6 +29428,12 @@ export interface components {
auto_router_model_compression?: string | null;
/** Auto Router Routing Compression */
auto_router_routing_compression?: string | null;
/** Auto Router Shunt Bulk Read Model */
auto_router_shunt_bulk_read_model?: string | null;
/** Auto Router Shunt Code Write Model */
auto_router_shunt_code_write_model?: string | null;
/** Auto Router Shunt Min Lines */
auto_router_shunt_min_lines?: number | null;
/** Aws Access Key Id */
aws_access_key_id?: string | null;
/** Aws Batch Role Arn */
@ -39477,6 +39605,12 @@ export interface components {
auto_router_model_compression?: string | null;
/** Auto Router Routing Compression */
auto_router_routing_compression?: string | null;
/** Auto Router Shunt Bulk Read Model */
auto_router_shunt_bulk_read_model?: string | null;
/** Auto Router Shunt Code Write Model */
auto_router_shunt_code_write_model?: string | null;
/** Auto Router Shunt Min Lines */
auto_router_shunt_min_lines?: number | null;
/** Aws Access Key Id */
aws_access_key_id?: string | null;
/** Aws Batch Role Arn */
@ -42477,6 +42611,41 @@ export interface operations {
};
};
};
bulk_read_bulk_read_post: {
parameters: {
query: {
router: string;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody: {
content: {
"multipart/form-data": components["schemas"]["Body_bulk_read_bulk_read_post"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": string;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
cache_delete_cache_delete_post: {
parameters: {
query?: never;
@ -43335,6 +43504,41 @@ export interface operations {
};
};
};
code_write_code_write_post: {
parameters: {
query: {
router: string;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody: {
content: {
"multipart/form-data": components["schemas"]["Body_code_write_code_write_post"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": string;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
cohere_proxy_route_cohere__endpoint__get: {
parameters: {
query?: never;
@ -61313,6 +61517,41 @@ export interface operations {
};
};
};
bulk_read_v1_bulk_read_post: {
parameters: {
query: {
router: string;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody: {
content: {
"multipart/form-data": components["schemas"]["Body_bulk_read_v1_bulk_read_post"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": string;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
chat_completion_v1_chat_completions_post: {
parameters: {
query?: never;
@ -61512,6 +61751,41 @@ export interface operations {
};
};
};
code_write_v1_code_write_post: {
parameters: {
query: {
router: string;
};
header?: never;
path?: never;
cookie?: never;
};
requestBody: {
content: {
"multipart/form-data": components["schemas"]["Body_code_write_v1_code_write_post"];
};
};
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": string;
};
};
/** @description Validation Error */
422: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": components["schemas"]["HTTPValidationError"];
};
};
};
};
completion_v1_completions_post: {
parameters: {
query?: {

2
ui/litellm-dashboard/tsconfig.tsbuildinfo
View file

File diff suppressed because one or more lines are too long