diff --git a/.github/PULL_REQUEST_TEMPLATE/general.md b/.github/PULL_REQUEST_TEMPLATE/general.md
index 4f1a18e925e..d766897ab3c 100644
--- a/.github/PULL_REQUEST_TEMPLATE/general.md
+++ b/.github/PULL_REQUEST_TEMPLATE/general.md
@@ -6,19 +6,12 @@
## Handwritten 1-3 sentence description
-
-``
-
-``
+[insert handwritten description here]
Details
diff --git a/.github/workflows/test-unit.yml b/.github/workflows/test-unit.yml
index 94152aeb7a0..f1456b1fde8 100644
--- a/.github/workflows/test-unit.yml
+++ b/.github/workflows/test-unit.yml
@@ -419,6 +419,7 @@ jobs:
- shard: router
test-path: >-
+ tests/unit/router_rust
tests/unit/test_router
python-version: "3.12"
workers: 4
diff --git a/cookbook/skill_scanner/README.md b/cookbook/skill_scanner/README.md
new file mode 100644
index 00000000000..2a68e116002
--- /dev/null
+++ b/cookbook/skill_scanner/README.md
@@ -0,0 +1,60 @@
+# Manual skill security scans
+
+This standalone adapter runs NVIDIA SkillSpector on a separate machine. It is not installed in the LiteLLM proxy and does not require a shared Helm or Compose deployment
+
+Start it with Python 3.12 or newer and uv. The script pins SkillSpector and its HTTP dependencies
+
+```sh
+export SKILL_SCANNER_API_KEY="$(openssl rand -hex 32)"
+uv run server.py
+```
+
+For another host, place the loopback service behind an HTTPS reverse proxy that forwards the Authorization header and enforces a 65 MiB request limit. Run it under an unprivileged account with a writable temporary directory and no production files or credentials. The scanner host needs Git and network access to requested sources and optional model providers. Apply network egress restrictions suitable for those sources
+
+Configure the same scanner secret in the gateway's environment, then add this to LiteLLM's configuration
+
+```yaml
+general_settings:
+ skill_scanner:
+ endpoint: https://scanner.example.com/scan
+ api_key_env: SKILL_SCANNER_API_KEY
+ use_llm: false
+ timeout_seconds: 180
+```
+
+The endpoint must be reachable from the gateway. A customer's laptop works while it is reachable through their network, VPN or authenticated HTTPS tunnel. `localhost` refers to the gateway machine, not the admin's browser. HTTP is accepted only for loopback development
+
+Open Skills, Submissions, and Review as a proxy admin. Click **Scan for threats**, inspect the report, then independently approve or reject the proposal. Scanning is optional and never changes approval permissions or blocks publication. The report shows findings, locations, remediation, risk score, scanner version and whether semantic analysis actually ran. Incomplete reports do not establish safety
+
+Uploaded ZIPs are sent privately to the scanner with their exact stored bytes. Metadata-only edits reuse the matching stored archive. If that archive is missing or changed, submit a new ZIP before scanning. The displayed SHA-256 identifies that archive, and the scan examines its files rather than proposed catalog metadata. Remote URL reports describe the source at scan time and do not freeze its contents. GitHub branch/tag refs and explicit-ref subdirectories are supported. Other ref/subdirectory combinations require a ZIP so the adapter does not silently scan the wrong revision
+
+The latest scan result is retained with the submission. Running scans have a deadline. If a gateway restarts or the scanner becomes unavailable, the attempt expires and an admin can retry. Retries replace the latest report. Approving or rejecting during a scan does not cancel it or change the decision
+
+Set `use_llm: true` to request semantic analysis and configure the provider on the scanner machine using [SkillSpector's provider instructions](https://github.com/NVIDIA/skillspector#llm-analysis). Provider credentials stay on that machine. Static analysis can also call OSV for dependency checks
+
+## HTTP contract
+
+The gateway sends an authenticated `POST` to the configured endpoint. The body contains form fields `source` (JSON source descriptor), `use_llm` (boolean), and `expires_at` (timezone-aware ISO timestamp), plus `file` (ZIP) when bytes are available. The service returns HTTP 200 with this JSON shape
+
+```json
+{
+ "scanner_version": "2.12.0",
+ "complete": true,
+ "scan_mode": "static-only",
+ "risk_score": 70,
+ "findings": [{
+ "rule_id": "example-rule",
+ "severity": "HIGH",
+ "message": "Potential data exfiltration",
+ "file": "SKILL.md",
+ "start_line": 8,
+ "explanation": "Instructions request sending local files to an external host",
+ "remediation": "Remove the external upload instruction"
+ }],
+ "limitations": []
+}
+```
+
+Reports are limited to 1 MiB and 500 findings. Scanner uploads allow 64 MiB because the gateway can expand compressed skill uploads into stored compatibility ZIPs. The gateway rejects malformed reports, refuses redirects and verifies TLS. Only the configured scanner credential is sent. A timeout, HTTP error or incomplete report remains visible to the reviewer
+
+The bundled adapter runs at most two child scans concurrently. Each child and its descendants are terminated at the gateway deadline, capped at 570 seconds. Keep the gateway and scanner clocks synchronized. Direct adapter clients may omit `expires_at` to use the cap. Child processes never execute submitted skill scripts. Raise the gateway deadline up to 600 seconds when needed for semantic analysis and configure the reverse proxy timeout to match
diff --git a/cookbook/skill_scanner/server.py b/cookbook/skill_scanner/server.py
new file mode 100644
index 00000000000..44a1cecb00f
--- /dev/null
+++ b/cookbook/skill_scanner/server.py
@@ -0,0 +1,172 @@
+# /// script
+# requires-python = ">=3.12"
+# dependencies = [
+# "fastapi==0.162.0", "uvicorn==0.54.0", "python-multipart==0.0.33",
+# "skillspector @ git+https://github.com/NVIDIA/skillspector.git@62c2fe8622f580ac292d4078c3b71f790194dbbd",
+# ]
+# ///
+import asyncio
+import hmac
+import json
+import os
+import signal
+import sys
+import tempfile
+from contextlib import suppress
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Annotated, Final
+from urllib.parse import quote, urlsplit
+
+from fastapi import Depends, FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from pydantic import AwareDatetime, TypeAdapter, ValidationError
+
+MAX_UPLOAD: Final = 64 * 1024 * 1024
+MAX_REPORT: Final = 1024 * 1024
+SCAN_TIMEOUT: Final = 570
+SOURCE: Final = TypeAdapter(dict[str, str])
+SLOTS: Final = asyncio.Semaphore(2)
+app: Final = FastAPI(title="SkillSpector adapter", docs_url=None, redoc_url=None, openapi_url=None)
+
+
+def authenticate(credentials: Annotated[HTTPAuthorizationCredentials, Depends(HTTPBearer())]) -> None:
+ token: Final = os.environ.get("SKILL_SCANNER_API_KEY", "")
+ if not token or not hmac.compare_digest(credentials.credentials, token):
+ raise HTTPException(401, "Invalid scanner credentials")
+
+
+def source_target(source: dict[str, str]) -> str:
+ target: Final = (
+ "https://github.com/" + source.get("repo", "") if source.get("source") == "github" else source.get("url", "")
+ )
+ parsed: Final = urlsplit(target)
+ if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
+ raise HTTPException(400, "Remote sources require an HTTPS URL without credentials or a fragment")
+ ref: Final = source.get("ref")
+ subdir: Final = source.get("path")
+ if ref or subdir:
+ if parsed.hostname != "github.com" or len(parsed.path.strip("/").split("/")) != 2 or not ref:
+ raise HTTPException(400, "Upload a ZIP to scan this source's exact ref and subdirectory")
+ suffix: Final = "/" + quote(subdir, safe="/") if subdir else ""
+ return target.removesuffix(".git").rstrip("/") + "/tree/" + quote(ref, safe="/") + suffix
+ return target
+
+
+async def worker(target: str, output: Path, use_llm: bool) -> None:
+ from skillspector.mcp_server import run_scan
+
+ result: Final = await run_scan(target, use_llm=use_llm, allow_local_targets=not target.startswith("https://"))
+ coverage: Final = result.get("analysis_completeness", {})
+ findings: Final = result["findings"]
+ complete: Final = (
+ result.get("execution_successful") is True
+ and coverage.get("is_complete") is True
+ and (not use_llm or (result.get("llm_available") is True and result.get("llm_used") is True))
+ and len(findings) <= 500
+ )
+ limitations: Final = tuple(str(value)[:2000] for value in coverage.get("limitations", ()))
+ report: Final = {
+ "scanner_version": result["version"],
+ "complete": complete,
+ "scan_mode": result["scan_mode"],
+ "risk_score": result["risk_score"],
+ "findings": [
+ {
+ "rule_id": finding["id"],
+ "severity": finding["severity"],
+ "message": finding.get("pattern") or finding.get("finding") or "Security finding",
+ "file": finding["location"]["file"],
+ "start_line": finding["location"].get("start_line"),
+ "explanation": finding.get("explanation"),
+ "remediation": finding.get("remediation"),
+ }
+ for finding in findings[:500]
+ ],
+ "limitations": (
+ *limitations,
+ *(("Only the first 500 findings are displayed.",) if len(findings) > 500 else ()),
+ ),
+ }
+ encoded: Final = json.dumps(report).encode()
+ if len(encoded) > MAX_REPORT:
+ raise ValueError("Scanner report exceeded the response limit")
+ output.write_bytes(encoded)
+
+
+async def run_worker(target: str, output: Path, use_llm: bool, expires_at: datetime | None = None) -> dict[str, object]:
+ timeout: Final = (
+ SCAN_TIMEOUT
+ if expires_at is None
+ else min(SCAN_TIMEOUT, (expires_at - datetime.now(timezone.utc)).total_seconds())
+ )
+ if timeout <= 0:
+ raise HTTPException(504, "SkillSpector scan deadline expired")
+ process: Final = await asyncio.create_subprocess_exec(
+ sys.executable,
+ str(Path(__file__).resolve()),
+ "--scan",
+ target,
+ str(output),
+ str(use_llm).lower(),
+ stdout=asyncio.subprocess.DEVNULL,
+ stderr=asyncio.subprocess.DEVNULL,
+ start_new_session=True,
+ env={
+ **os.environ,
+ "HOME": str(output.parent),
+ "TMPDIR": str(output.parent),
+ "GIT_CONFIG_GLOBAL": os.devnull,
+ "GIT_CONFIG_NOSYSTEM": "1",
+ "GIT_TERMINAL_PROMPT": "0",
+ "GIT_ALLOW_PROTOCOL": "https",
+ },
+ )
+ try:
+ code: Final = await asyncio.wait_for(process.wait(), timeout)
+ if code != 0 or not output.is_file() or output.stat().st_size > MAX_REPORT:
+ raise HTTPException(502, "SkillSpector could not produce a report for this source")
+ return TypeAdapter(dict[str, object]).validate_json(output.read_bytes())
+ except TimeoutError:
+ raise HTTPException(504, "SkillSpector scan timed out") from None
+ finally:
+ with suppress(ProcessLookupError):
+ os.killpg(process.pid, signal.SIGKILL)
+ await process.wait()
+
+
+@app.post("/scan", dependencies=[Depends(authenticate)])
+async def scan(
+ source: Annotated[str, Form(max_length=16384)],
+ use_llm: Annotated[bool, Form()] = False,
+ expires_at: Annotated[AwareDatetime | None, Form()] = None,
+ file: Annotated[UploadFile | None, File()] = None,
+) -> dict[str, object]:
+ if SLOTS.locked():
+ raise HTTPException(429, "Scanner is busy; try again shortly")
+ try:
+ parsed: Final = SOURCE.validate_json(source)
+ except ValidationError:
+ raise HTTPException(400, "Invalid source metadata") from None
+ async with SLOTS:
+ with tempfile.TemporaryDirectory(prefix="skill-scan-") as directory:
+ root: Final = Path(directory)
+ if file is not None:
+ content: Final = await file.read(MAX_UPLOAD + 1)
+ if len(content) > MAX_UPLOAD:
+ raise HTTPException(413, "Upload exceeds 64 MiB")
+ archive: Final = root / "skill.zip"
+ archive.write_bytes(content)
+ return await run_worker(str(archive), root / "report.json", use_llm, expires_at)
+ return await run_worker(source_target(parsed), root / "report.json", use_llm, expires_at)
+
+
+if __name__ == "__main__":
+ if len(sys.argv) > 1 and sys.argv[1] == "--scan":
+ asyncio.run(worker(sys.argv[2], Path(sys.argv[3]), sys.argv[4] == "true"))
+ else:
+ import uvicorn
+
+ if not os.environ.get("SKILL_SCANNER_API_KEY"):
+ raise SystemExit("Set SKILL_SCANNER_API_KEY before starting the scanner")
+ uvicorn.run(app, host="127.0.0.1", port=int(os.environ.get("PORT", "8099")))
diff --git a/enterprise/litellm_enterprise/types/enterprise_callbacks/send_emails.py b/enterprise/litellm_enterprise/types/enterprise_callbacks/send_emails.py
index a2322d7f650..bb3b712d50b 100644
--- a/enterprise/litellm_enterprise/types/enterprise_callbacks/send_emails.py
+++ b/enterprise/litellm_enterprise/types/enterprise_callbacks/send_emails.py
@@ -1,9 +1,9 @@
import enum
-from typing import Dict, List, Optional
from pydantic import Field
from litellm.proxy._types import WebhookEvent
+from litellm.types.integrations.model_deprecation_email import MODEL_DEPRECATION_EMAIL_EVENT
from litellm.types.llms.base import LiteLLMBaseModel
@@ -26,7 +26,7 @@ class SendKeyCreatedEmailEvent(WebhookEvent):
class SendKeyRotatedEmailEvent(WebhookEvent):
virtual_key: str
- key_alias: Optional[str] = None
+ key_alias: str | None = None
"""
The virtual key that was rotated
this will be sk-123xxx, since we will be emailing this to the user to start using the new key
@@ -34,23 +34,33 @@ class SendKeyRotatedEmailEvent(WebhookEvent):
class EmailEvent(str, enum.Enum):
+ model_deprecations = MODEL_DEPRECATION_EMAIL_EVENT
virtual_key_created = "Virtual Key Created"
new_user_invitation = "New User Invitation"
virtual_key_rotated = "Virtual Key Rotated"
soft_budget_crossed = "Soft Budget Crossed"
max_budget_alert = "Max Budget Alert"
+
class EmailEventSettings(LiteLLMBaseModel):
event: EmailEvent
enabled: bool
+
+
class EmailEventSettingsUpdateRequest(LiteLLMBaseModel):
- settings: List[EmailEventSettings]
+ settings: list[EmailEventSettings]
+
+
class EmailEventSettingsResponse(LiteLLMBaseModel):
- settings: List[EmailEventSettings]
+ settings: list[EmailEventSettings]
+
+
class DefaultEmailSettings(LiteLLMBaseModel):
"""Default settings for email events"""
- settings: Dict[EmailEvent, bool] = Field(
+
+ settings: dict[EmailEvent, bool] = Field(
default_factory=lambda: {
+ EmailEvent.model_deprecations: False, # Explicit opt-in for owner retirement reminders.
EmailEvent.virtual_key_created: True, # On by default
EmailEvent.new_user_invitation: True, # On by default
EmailEvent.virtual_key_rotated: True, # On by default
@@ -58,10 +68,12 @@ class DefaultEmailSettings(LiteLLMBaseModel):
EmailEvent.max_budget_alert: True, # On by default
}
)
- def to_dict(self) -> Dict[str, bool]:
+
+ def to_dict(self) -> dict[str, bool]:
"""Convert to dictionary with string keys for storage"""
return {event.value: enabled for event, enabled in self.settings.items()}
+
@classmethod
- def get_defaults(cls) -> Dict[str, bool]:
+ def get_defaults(cls) -> dict[str, bool]:
"""Get the default settings as a dictionary with string keys"""
- return cls().to_dict()
\ No newline at end of file
+ return cls().to_dict()
diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261103010000_spendlogs_accounting_delivery/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261103010000_spendlogs_accounting_delivery/migration.sql
new file mode 100644
index 00000000000..b00cf892b68
--- /dev/null
+++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261103010000_spendlogs_accounting_delivery/migration.sql
@@ -0,0 +1,19 @@
+SET LOCAL lock_timeout = '10s';
+CREATE TABLE IF NOT EXISTS "LiteLLM_SpendLogDelivery" (
+ "delivery_id" TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
+ "request_id" TEXT NOT NULL,
+ "state" "AccountingDeliveryState" NOT NULL DEFAULT 'pending',
+ "attempts" INTEGER NOT NULL DEFAULT 0,
+ "next_attempt_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ "applied_at" TIMESTAMP(3),
+ "last_error" TEXT,
+ "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ "accounting_payload" JSONB NOT NULL,
+ "is_batch" BOOLEAN NOT NULL
+);
+CREATE INDEX IF NOT EXISTS "LiteLLM_SpendLogDelivery_state_next_attempt_at_idx"
+ ON "LiteLLM_SpendLogDelivery" ("state", "next_attempt_at");
+CREATE INDEX IF NOT EXISTS "LiteLLM_SpendLogDelivery_state_applied_at_idx"
+ ON "LiteLLM_SpendLogDelivery" ("state", "applied_at");
+CREATE INDEX IF NOT EXISTS "LiteLLM_SpendLogDelivery_request_id_idx"
+ ON "LiteLLM_SpendLogDelivery" ("request_id");
diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261104010000_drop_accounting_event_tables/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261104010000_drop_accounting_event_tables/migration.sql
new file mode 100644
index 00000000000..cba6a18368f
--- /dev/null
+++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261104010000_drop_accounting_event_tables/migration.sql
@@ -0,0 +1,5 @@
+SET LOCAL lock_timeout = '10s';
+DROP TABLE IF EXISTS "LiteLLM_AccountingDelivery";
+DROP TABLE IF EXISTS "LiteLLM_AccountingEvent";
+DROP TYPE IF EXISTS "AccountingDeliveryTarget";
+DROP TYPE IF EXISTS "AccountingEventKind";
diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_skill_version_history/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_skill_version_history/migration.sql
new file mode 100644
index 00000000000..ecfa7321d87
--- /dev/null
+++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105000000_skill_version_history/migration.sql
@@ -0,0 +1,16 @@
+SET LOCAL lock_timeout = '10s';
+CREATE TABLE IF NOT EXISTS "LiteLLM_ClaudeCodePluginVersion" (
+ "id" TEXT NOT NULL,
+ "plugin_id" TEXT NOT NULL,
+ "version" TEXT,
+ "description" TEXT,
+ "manifest_json" TEXT,
+ "published_at" TIMESTAMP(3),
+ "published_by" TEXT,
+ "archive_available" BOOLEAN NOT NULL DEFAULT false,
+ "content" BYTEA,
+ CONSTRAINT "LiteLLM_ClaudeCodePluginVersion_pkey" PRIMARY KEY ("id"),
+ CONSTRAINT "LiteLLM_ClaudeCodePluginVersion_plugin_id_fkey" FOREIGN KEY ("plugin_id") REFERENCES "LiteLLM_ClaudeCodePluginTable"("id") ON DELETE CASCADE ON UPDATE CASCADE
+);
+CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_ClaudeCodePluginVersion_plugin_id_version_key" ON "LiteLLM_ClaudeCodePluginVersion"("plugin_id", "version");
+CREATE INDEX IF NOT EXISTS "LiteLLM_ClaudeCodePluginVersion_plugin_id_published_at_idx" ON "LiteLLM_ClaudeCodePluginVersion"("plugin_id", "published_at");
diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105010000_skill_submissions/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105010000_skill_submissions/migration.sql
new file mode 100644
index 00000000000..e7975258ca9
--- /dev/null
+++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105010000_skill_submissions/migration.sql
@@ -0,0 +1,27 @@
+SET LOCAL lock_timeout = '10s';
+
+CREATE TABLE IF NOT EXISTS "LiteLLM_SkillSubmission" (
+ "id" TEXT NOT NULL,
+ "plugin_name" TEXT NOT NULL,
+ "operation" TEXT NOT NULL,
+ "status" TEXT NOT NULL DEFAULT 'pending_review',
+ "proposal_json" TEXT NOT NULL,
+ "archive_content" BYTEA,
+ "submitted_by" TEXT NOT NULL,
+ "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ "base_plugin_id" TEXT,
+ "base_updated_at" TIMESTAMP(3),
+ "reviewed_by" TEXT,
+ "reviewed_at" TIMESTAMP(3),
+ "rejection_reason" TEXT,
+ CONSTRAINT "LiteLLM_SkillSubmission_pkey" PRIMARY KEY ("id")
+);
+
+CREATE INDEX IF NOT EXISTS "LiteLLM_SkillSubmission_submitted_by_created_at_id_idx"
+ ON "LiteLLM_SkillSubmission"("submitted_by", "created_at" DESC, "id" DESC);
+CREATE INDEX IF NOT EXISTS "LiteLLM_SkillSubmission_submitted_by_status_created_at_id_idx"
+ ON "LiteLLM_SkillSubmission"("submitted_by", "status", "created_at" DESC, "id" DESC);
+CREATE INDEX IF NOT EXISTS "LiteLLM_SkillSubmission_status_created_at_id_idx"
+ ON "LiteLLM_SkillSubmission"("status", "created_at" DESC, "id" DESC);
+CREATE INDEX IF NOT EXISTS "LiteLLM_SkillSubmission_created_at_id_idx"
+ ON "LiteLLM_SkillSubmission"("created_at" DESC, "id" DESC);
diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105020000_skill_scanning/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105020000_skill_scanning/migration.sql
new file mode 100644
index 00000000000..6b70cca4a9a
--- /dev/null
+++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105020000_skill_scanning/migration.sql
@@ -0,0 +1,3 @@
+SET LOCAL lock_timeout = '10s';
+
+ALTER TABLE "LiteLLM_SkillSubmission" ADD COLUMN IF NOT EXISTS "scan_json" TEXT;
diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105094500_add_skill_download_history/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105094500_add_skill_download_history/migration.sql
new file mode 100644
index 00000000000..43e26b324a0
--- /dev/null
+++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20261105094500_add_skill_download_history/migration.sql
@@ -0,0 +1,37 @@
+SET LOCAL lock_timeout = '10s';
+
+DO $$
+BEGIN
+ IF NOT EXISTS (
+ SELECT 1 FROM pg_attribute
+ WHERE attrelid = to_regclass('"LiteLLM_ClaudeCodePluginTable"')
+ AND attname = 'anonymous_download_count' AND NOT attisdropped
+ ) THEN
+ ALTER TABLE "LiteLLM_ClaudeCodePluginTable"
+ ADD COLUMN IF NOT EXISTS "anonymous_download_count" BIGINT NOT NULL DEFAULT 0;
+ END IF;
+
+ IF to_regclass('"LiteLLM_ClaudeCodePluginDownload"') IS NULL THEN
+ CREATE TABLE IF NOT EXISTS "LiteLLM_ClaudeCodePluginDownload" (
+ "id" BIGSERIAL NOT NULL,
+ "plugin_id" TEXT NOT NULL,
+ "downloaded_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ "version" TEXT,
+ "digest" TEXT NOT NULL,
+ "authenticated" BOOLEAN NOT NULL,
+ "user_id" TEXT,
+ "user_email" TEXT,
+ "key_alias" TEXT,
+ "team_id" TEXT,
+ "team_alias" TEXT,
+ CONSTRAINT "LiteLLM_ClaudeCodePluginDownload_pkey" PRIMARY KEY ("id"),
+ CONSTRAINT "LiteLLM_ClaudeCodePluginDownload_plugin_id_fkey" FOREIGN KEY ("plugin_id")
+ REFERENCES "LiteLLM_ClaudeCodePluginTable"("id") ON DELETE CASCADE ON UPDATE CASCADE
+ );
+ END IF;
+
+ IF to_regclass('"LiteLLM_ClaudeCodePluginDownload_plugin_id_id_idx"') IS NULL THEN
+ CREATE INDEX IF NOT EXISTS "LiteLLM_ClaudeCodePluginDownload_plugin_id_id_idx"
+ ON "LiteLLM_ClaudeCodePluginDownload"("plugin_id", "id");
+ END IF;
+END $$;
diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma
index c007b047087..818f2dc4858 100644
--- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma
+++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma
@@ -1616,10 +1616,69 @@ model LiteLLM_ClaudeCodePluginTable {
created_by String?
uploaded_archive LiteLLM_ClaudeCodePluginUpload?
+ versions LiteLLM_ClaudeCodePluginVersion[]
+ downloads LiteLLM_ClaudeCodePluginDownload[]
+ anonymous_download_count BigInt @default(0)
@@map("LiteLLM_ClaudeCodePluginTable")
}
+model LiteLLM_ClaudeCodePluginVersion {
+ id String @id @default(uuid())
+ plugin_id String
+ version String?
+ description String?
+ manifest_json String?
+ published_at DateTime?
+ published_by String?
+ archive_available Boolean @default(false)
+ content Bytes?
+ plugin LiteLLM_ClaudeCodePluginTable @relation(fields: [plugin_id], references: [id], onDelete: Cascade, onUpdate: Cascade)
+
+ @@unique([plugin_id, version])
+ @@index([plugin_id, published_at])
+}
+
+
+model LiteLLM_SkillSubmission {
+ id String @id @default(uuid())
+ plugin_name String
+ operation String
+ status String @default("pending_review")
+ proposal_json String
+ archive_content Bytes?
+ submitted_by String
+ created_at DateTime @default(now())
+ base_plugin_id String?
+ base_updated_at DateTime?
+ reviewed_by String?
+ reviewed_at DateTime?
+ rejection_reason String?
+ scan_json String?
+
+ @@index([submitted_by, created_at(sort: Desc), id(sort: Desc)])
+ @@index([submitted_by, status, created_at(sort: Desc), id(sort: Desc)])
+ @@index([status, created_at(sort: Desc), id(sort: Desc)])
+ @@index([created_at(sort: Desc), id(sort: Desc)])
+}
+
+model LiteLLM_ClaudeCodePluginDownload {
+ id BigInt @id @default(autoincrement())
+ plugin_id String
+ downloaded_at DateTime @default(now())
+ version String?
+ digest String
+ authenticated Boolean
+ user_id String?
+ user_email String?
+ key_alias String?
+ team_id String?
+ team_alias String?
+ plugin LiteLLM_ClaudeCodePluginTable @relation(fields: [plugin_id], references: [id], onDelete: Cascade, onUpdate: Cascade)
+
+ @@index([plugin_id, id])
+}
+
model LiteLLM_ClaudeCodePluginUpload {
plugin_name String @id
content Bytes
@@ -2071,74 +2130,27 @@ model LiteLLM_TelemetryReport {
@@index([created_at])
}
-enum AccountingEventKind {
- inbound_usage
- outbound_usage
- cost
-}
-
enum AccountingDeliveryState {
pending
applied
parked
}
-enum AccountingDeliveryTarget {
- LiteLLM_VerificationToken
- LiteLLM_UserTable
- LiteLLM_TeamTable
- LiteLLM_TeamMembership
- LiteLLM_OrganizationTable
- LiteLLM_OrganizationMembership
- LiteLLM_EndUserTable
- LiteLLM_AgentsTable
- LiteLLM_TagTable
- LiteLLM_ProjectTable
- LiteLLM_ModelAccessGroupBudgetTable
- LiteLLM_BudgetTable
- LiteLLM_BudgetWindowSpend
- LiteLLM_DailyUserSpend
- LiteLLM_DailyTeamSpend
- LiteLLM_DailyOrganizationSpend
- LiteLLM_DailyEndUserSpend
- LiteLLM_DailyAgentSpend
- LiteLLM_DailyTagSpend
- LiteLLM_DailyModelUsage
- LiteLLM_DailyToolSpend
- LiteLLM_AutoRouterDailySpend
-}
+model LiteLLM_SpendLogDelivery {
+ delivery_id String @id @default(dbgenerated("(gen_random_uuid())::text"))
+ request_id String
+ state AccountingDeliveryState @default(pending)
+ accounting_payload Json
+ is_batch Boolean
+ attempts Int @default(0)
+ next_attempt_at DateTime @default(now())
+ applied_at DateTime?
+ last_error String?
+ created_at DateTime @default(now())
-// Durable accounting ledger. One row per usage or cost event; payload holds the
-// private event envelope versioned by schema_version. Usage rows retain request-scoped
-// pricing context; cost rows include resolved rates and the finalized pricing snapshot.
-model LiteLLM_AccountingEvent {
- event_id String @id
- kind AccountingEventKind
- source_event_id String?
- schema_version Int
- payload_hash String
- occurred_at DateTime
- recorded_at DateTime @default(now())
- payload Json
-
- @@index([recorded_at])
- @@index([kind, source_event_id])
-}
-
-// One row per (event, target table, version): pending work for the projector and,
-// once applied in the same transaction as the target write, the receipt that stops a retry.
-model LiteLLM_AccountingDelivery {
- event_id String
- target AccountingDeliveryTarget
- version Int
- state AccountingDeliveryState @default(pending)
- attempts Int @default(0)
- next_attempt_at DateTime @default(now())
- applied_at DateTime?
- last_error String?
-
- @@id([event_id, target, version])
- @@index([target, state, next_attempt_at])
+ @@index([state, next_attempt_at])
+ @@index([state, applied_at])
+ @@index([request_id])
}
model LiteLLM_SubscriptionAccountTable {
diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock
index ba1131d7a92..23ebfc7d3cf 100644
--- a/litellm-rust/Cargo.lock
+++ b/litellm-rust/Cargo.lock
@@ -4288,6 +4288,7 @@ dependencies = [
name = "litellm-python-bridge"
version = "0.1.0"
dependencies = [
+ "async-trait",
"aws-sdk-secretsmanager",
"bytes",
"criterion",
@@ -4312,6 +4313,7 @@ dependencies = [
"litellm-llms",
"litellm-llms-types",
"litellm-model-catalog",
+ "litellm-router-core",
"litellm-router-types",
"litellm-secret-value",
"litellm-secrets",
@@ -4393,6 +4395,24 @@ dependencies = [
[[package]]
name = "litellm-router-core"
version = "0.1.0"
+dependencies = [
+ "arc-swap",
+ "async-trait",
+ "jiff",
+ "litellm-cache",
+ "litellm-router-types",
+ "parking_lot",
+ "rand 0.8.7",
+ "rstest",
+ "serde",
+ "serde_json",
+ "serde_with",
+ "sha2 0.10.9",
+ "strum",
+ "thiserror 2.0.19",
+ "tokio",
+ "tracing",
+]
[[package]]
name = "litellm-router-types"
diff --git a/litellm-rust/Cargo.toml b/litellm-rust/Cargo.toml
index 943e8576ae6..71914a0bffe 100644
--- a/litellm-rust/Cargo.toml
+++ b/litellm-rust/Cargo.toml
@@ -13,6 +13,7 @@ litellm-config = { path = "crates/config" }
litellm-error = { path = "crates/error" }
litellm-router = { path = "crates/router" }
litellm-router-types = { path = "crates/router-types" }
+litellm-router-core = { path = "crates/router-core" }
litellm-tracing = { path = "crates/tracing" }
litellm-spend-clickhouse = { path = "crates/spend-clickhouse" }
litellm-storage-clickhouse = { path = "crates/storage-clickhouse" }
@@ -76,6 +77,8 @@ litellm-cost = { path = "crates/cost" }
litellm-model-catalog = { path = "crates/model-catalog" }
litellm-model-catalog-refresh = { path = "crates/model-catalog-refresh" }
+arc-swap = "1.9.2"
+async-trait = "0.1.91"
askama = { version = "0.16.1", default-features = false, features = ["derive", "std"] }
tracing = "0.1"
axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio", "multipart"] }
diff --git a/litellm-rust/crates/python-bridge/Cargo.toml b/litellm-rust/crates/python-bridge/Cargo.toml
index fd826f549e5..ba892199dcf 100644
--- a/litellm-rust/crates/python-bridge/Cargo.toml
+++ b/litellm-rust/crates/python-bridge/Cargo.toml
@@ -25,6 +25,7 @@ litellm-tracing.workspace = true
litellm-spend-clickhouse.workspace = true
litellm-storage-clickhouse.workspace = true
litellm-host.workspace = true
+async-trait.workspace = true
bytes.workspace = true
futures-util.workspace = true
litellm-cache.workspace = true
@@ -47,6 +48,7 @@ litellm-secrets = { workspace = true, features = ["aws", "azure", "google", "has
litellm-secrets-types.workspace = true
litellm-llms-types.workspace = true
litellm-router-types.workspace = true
+litellm-router-core.workspace = true
litellm-host-python.workspace = true
litellm-model-catalog.workspace = true
litellm-token-counter = { path = "../token-counter", default-features = false }
diff --git a/litellm-rust/crates/python-bridge/src/lib.rs b/litellm-rust/crates/python-bridge/src/lib.rs
index de35419308a..f4a4f7c2b26 100644
--- a/litellm-rust/crates/python-bridge/src/lib.rs
+++ b/litellm-rust/crates/python-bridge/src/lib.rs
@@ -12,6 +12,7 @@ mod marshal;
mod model_registry;
mod preflight;
mod python_settings;
+mod router;
mod routes;
mod secrets;
mod tokenizer;
@@ -36,6 +37,8 @@ mod _native {
#[pymodule_export]
use crate::model_registry::bundled_model_cost_map;
#[pymodule_export]
+ use crate::router::{RouterCall, RouterCoreHandle};
+ #[pymodule_export]
use crate::routes::audio_transcription::{atranscription, transcription};
#[pymodule_export]
use crate::routes::chat_completions::{acompletion, completion};
@@ -115,6 +118,8 @@ mod tests {
"bundled_model_cost_map",
"TokenCounter",
"Tokenizer",
+ "RouterCoreHandle",
+ "RouterCall",
"gil_stats",
"process_state_started",
"reserve_process_for_forking",
diff --git a/litellm-rust/crates/python-bridge/src/router/config.rs b/litellm-rust/crates/python-bridge/src/router/config.rs
new file mode 100644
index 00000000000..de383a0ffa1
--- /dev/null
+++ b/litellm-rust/crates/python-bridge/src/router/config.rs
@@ -0,0 +1,412 @@
+//! The documents `RouterCoreHandle` reads: the router's static and dynamic config, a call's
+//! request and an `apply` change set. A config key router-core does not read declines the
+//! handle, so Python keeps the router on its own core.
+
+use std::collections::BTreeMap;
+
+use litellm_router_core::select::PassThrough;
+use litellm_router_core::{
+ ChangeSet, ConfigError, DeploymentSpec, OutcomeCallbacks, RouterConfig, RouterDynamicConfig,
+ RouterRequest,
+ config::{Fallbacks, RetryPolicy},
+ host::UncountedRetries,
+ mock::MockFailure,
+ registry::IndexOverrides,
+};
+use serde::{Deserialize, de::DeserializeOwned};
+use serde_json::{Map, Value};
+
+/// The `RouterConfig` keys router-core reads.
+const STATIC_KEYS: &[&str] = &[
+ "model_list",
+ "routing_strategy",
+ "routing_groups",
+ "routing_group_members",
+ "plugins",
+ "cooldown_time",
+ "disable_cooldowns",
+ "allowed_fails_policy",
+ "retry_policy",
+ "model_group_retry_policy",
+ "retry_after",
+ "timeout",
+ "request_timeout",
+ "drop_params",
+ "enable_pre_call_checks",
+ "routing_plugins",
+ "callback_filters",
+ "enable_health_check_routing",
+ "health_check_staleness_threshold",
+ "background_health_check_model_groups",
+ "enable_weighted_failover",
+ "fallback_on_unsupported_params",
+ "provider_default_deployment_ids",
+];
+
+/// The `RouterDynamicConfig` keys router-core reads.
+const DYNAMIC_KEYS: &[&str] = &[
+ "version",
+ "model_alias_map",
+ "model_group_alias",
+ "fallbacks",
+ "default_fallbacks",
+ "context_window_fallbacks",
+ "content_policy_fallbacks",
+ "num_retries",
+ "max_fallbacks",
+ "allowed_fails",
+ "allowed_fails_is_default",
+ "fallback_keys",
+ "default_deployment",
+ "default_deployment_raises",
+];
+
+/// Router arguments that do not change routing, accepted in either config and dropped.
+const IGNORED_KEYS: &[&str] = &[
+ "set_verbose",
+ "debug_level",
+ "alerting_config",
+ "client_ttl",
+];
+
+/// Fallback lists Python treats as unset when empty (`context_window_fallbacks or None`).
+const EMPTY_MEANS_UNSET: &[&str] = &["context_window_fallbacks", "content_policy_fallbacks"];
+
+/// Which document a failure is about.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::Display)]
+pub(crate) enum Document {
+ #[strum(serialize = "router config")]
+ Static,
+ #[strum(serialize = "dynamic router config")]
+ Dynamic,
+ #[strum(serialize = "router request")]
+ Request,
+ #[strum(serialize = "change set")]
+ Changes,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum Error {
+ #[error("the {document} must be a JSON object")]
+ NotObject { document: Document },
+ #[error("router-core does not read `{key}` in the {document}")]
+ UnknownKey { document: Document, key: String },
+ #[error("the {document} does not have the shape router-core reads")]
+ Shape {
+ document: Document,
+ #[source]
+ source: serde_json::Error,
+ },
+ #[error("the request names no model")]
+ MissingModel,
+ #[error("router-core rejected the router config")]
+ Core(#[from] ConfigError),
+}
+
+impl Error {
+ /// Whether router-core cannot serve this config, so Python keeps its own core.
+ pub(crate) fn declined(&self) -> bool {
+ match self {
+ Self::UnknownKey { .. } | Self::Core(_) => true,
+ Self::Shape { document, .. } => {
+ matches!(document, Document::Static | Document::Dynamic)
+ }
+ Self::NotObject { .. } | Self::MissingModel => false,
+ }
+ }
+}
+
+fn object(document: Document, value: Value) -> Result, Error> {
+ match value {
+ Value::Object(object) => Ok(object),
+ _ => Err(Error::NotObject { document }),
+ }
+}
+
+fn shaped(document: Document, object: Map) -> Result {
+ serde_json::from_value(Value::Object(object))
+ .map_err(|source| Error::Shape { document, source })
+}
+
+/// Drop unset and ignored keys, then reject any key outside `known`.
+fn routing_keys(
+ document: Document,
+ value: Value,
+ known: &[&str],
+) -> Result, Error> {
+ let mut object = object(document, value)?;
+ object.retain(|key, value| {
+ let empty_unset = EMPTY_MEANS_UNSET.contains(&key.as_str())
+ && value.as_array().is_some_and(Vec::is_empty);
+ !(value.is_null() || empty_unset || IGNORED_KEYS.contains(&key.as_str()))
+ });
+ if let Some(key) = object.keys().find(|key| !known.contains(&key.as_str())) {
+ return Err(Error::UnknownKey {
+ document,
+ key: key.clone(),
+ });
+ }
+ Ok(object)
+}
+
+pub(crate) fn parse_router_config(value: Value) -> Result {
+ let object = routing_keys(Document::Static, value, STATIC_KEYS)?;
+ shaped(Document::Static, object)
+}
+
+pub(crate) fn parse_dynamic_config(value: Value) -> Result {
+ let object = routing_keys(Document::Dynamic, value, DYNAMIC_KEYS)?;
+ shaped(Document::Dynamic, object)
+}
+
+/// A call's request as Python spells it.
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RequestSpec {
+ model: Option,
+ team_id: Option,
+ listing_team_id: Option,
+ proxy_admin: Option,
+ allowed_models: Option>,
+ admin_model: Option,
+ specific_deployment: Option,
+ num_retries: Option,
+ uncounted_retries: Option,
+ disable_fallbacks: Option,
+ router_weights: Option,
+ pinned_deployment_id: Option,
+ routing_strategy: Option,
+ filter_tags: Option,
+ /// `listed` or `healthy`: which pass-through pick the request makes.
+ pass_through: Option,
+ target_order: Option,
+ excluded_ids: Option>,
+ retry_skipped_ids: Option>,
+ pre_routes: Option,
+ has_image: Option,
+ endpoint: Option,
+ web_search: Option,
+ route_handler: Option,
+ decisions_media: Option,
+ batch_create: Option,
+ embedding_batch: Option,
+ wildcard_request: Option,
+ unrouted: Option,
+ passthrough_on_no_deployment: Option,
+ skip_weighted_failover: Option,
+ same_group_fallbacks_only: Option,
+ model_group_retry_policy: Option>,
+ retry_policy_raises: Option>,
+ #[serde(default, deserialize_with = "present")]
+ fallbacks: Option>,
+ #[serde(default, deserialize_with = "present")]
+ context_window_fallbacks: Option>,
+ #[serde(default, deserialize_with = "present")]
+ content_policy_fallbacks: Option>,
+ max_fallbacks: Option,
+ mock_failure: Option,
+ cooldown_time: Option,
+ health_check_probe: Option,
+ outcome_callbacks: Option,
+ body: Option,
+}
+
+/// A key the request passed, even as `None`: `Some(None)`.
+fn present<'de, D, T>(deserializer: D) -> Result>, D::Error>
+where
+ D: serde::Deserializer<'de>,
+ T: Deserialize<'de>,
+{
+ Option::::deserialize(deserializer).map(Some)
+}
+
+/// Read a request; `model` replaces the request's own when given.
+pub(crate) fn parse_request(value: Value, model: Option<&str>) -> Result {
+ let spec: RequestSpec = shaped(Document::Request, object(Document::Request, value)?)?;
+ let model = model
+ .map(str::to_owned)
+ .or(spec.model)
+ .ok_or(Error::MissingModel)?;
+ Ok(RouterRequest {
+ model,
+ body: spec.body.unwrap_or_default(),
+ team_id: spec.team_id,
+ listing_team_id: spec.listing_team_id,
+ proxy_admin: spec.proxy_admin.unwrap_or(false),
+ allowed_models: spec.allowed_models.unwrap_or_default(),
+ admin_model: spec.admin_model,
+ specific_deployment: spec.specific_deployment.unwrap_or(false),
+ num_retries: spec.num_retries,
+ uncounted_retries: spec.uncounted_retries,
+ disable_fallbacks: spec.disable_fallbacks.unwrap_or(false),
+ router_weights: spec.router_weights,
+ pinned_deployment_id: spec.pinned_deployment_id,
+ routing_strategy: spec.routing_strategy,
+ filter_tags: spec.filter_tags.unwrap_or(false),
+ pass_through: match spec.pass_through.as_deref() {
+ None => None,
+ Some("healthy") => Some(PassThrough::Healthy),
+ Some(_) => Some(PassThrough::Listed),
+ },
+ target_order: spec.target_order,
+ excluded_ids: spec.excluded_ids.unwrap_or_default(),
+ retry_skipped_ids: spec.retry_skipped_ids.unwrap_or_default(),
+ pre_routes: spec.pre_routes.unwrap_or(false),
+ has_image: spec.has_image.unwrap_or(false),
+ endpoint: spec.endpoint,
+ web_search: spec.web_search.unwrap_or(false),
+ route_handler: spec.route_handler,
+ decisions_media: spec.decisions_media.unwrap_or(false),
+ batch_create: spec.batch_create.unwrap_or(false),
+ embedding_batch: spec.embedding_batch.unwrap_or(false),
+ wildcard_request: spec.wildcard_request,
+ unrouted: spec.unrouted.unwrap_or(false),
+ passthrough_on_no_deployment: spec.passthrough_on_no_deployment.unwrap_or(false),
+ skip_weighted_failover: spec.skip_weighted_failover.unwrap_or(false),
+ same_group_fallbacks_only: spec.same_group_fallbacks_only.unwrap_or(false),
+ model_group_retry_policy: spec.model_group_retry_policy,
+ retry_policy_raises: spec.retry_policy_raises.unwrap_or_default(),
+ fallbacks: spec.fallbacks,
+ context_window_fallbacks: spec.context_window_fallbacks,
+ content_policy_fallbacks: spec.content_policy_fallbacks,
+ max_fallbacks: spec.max_fallbacks,
+ mock_failure: spec.mock_failure,
+ cooldown_time: spec.cooldown_time,
+ health_check_probe: spec.health_check_probe.unwrap_or(false),
+ outcome_callbacks: spec.outcome_callbacks.unwrap_or_default(),
+ payload: None,
+ })
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ChangesSpec {
+ /// Every deployment, replacing the current list before the upserts and deletes.
+ #[serde(default)]
+ replace: Option>,
+ /// Deployments added at the end, beside any that share their id.
+ #[serde(default)]
+ append: Vec,
+ #[serde(default)]
+ upsert: Vec,
+ #[serde(default)]
+ delete: Vec,
+ /// A new dynamic document (the same keys `RouterCoreHandle(static, dynamic)` takes).
+ #[serde(default)]
+ settings: Option,
+ /// The registry indexes the host wrote (`IndexOverrides`); `{}` drops them.
+ #[serde(default)]
+ index: Option,
+ /// The deployment at the id map's position for this id, removed with every index shifted.
+ #[serde(default)]
+ remove: Option,
+ /// A write to the indexes the host hands out (`HostIndexes`), field by field.
+ #[serde(default)]
+ host_index: Option,
+}
+
+pub(crate) fn parse_changes(value: Value) -> Result {
+ let spec: ChangesSpec = shaped(Document::Changes, object(Document::Changes, value)?)?;
+ Ok(ChangeSet {
+ replace: spec.replace,
+ appends: spec.append,
+ upserts: spec.upsert,
+ deletes: spec.delete,
+ settings: spec.settings.map(parse_dynamic_config).transpose()?,
+ index: spec.index,
+ remove: spec.remove,
+ host_index: spec.host_index,
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use rstest::rstest;
+ use serde_json::json;
+
+ use super::*;
+
+ #[rstest]
+ #[case::unknown_key(json!({"routing_strategy_args": {"ttl": 1}}), "routing_strategy_args")]
+ #[case::dynamic_key_in_static(json!({"num_retries": 2}), "num_retries")]
+ fn an_unread_static_key_declines_naming_the_key(#[case] json: Value, #[case] key: &str) {
+ let error = parse_router_config(json).unwrap_err();
+ assert!(error.declined());
+ assert!(matches!(&error, Error::UnknownKey { key: named, .. } if named == key));
+ }
+
+ #[rstest]
+ #[case::null_unknown(json!({"routing_strategy_args": null}))]
+ #[case::ignored(json!({"set_verbose": true, "debug_level": "INFO"}))]
+ #[case::pre_call_checks_off(json!({"enable_pre_call_checks": false}))]
+ #[case::pre_call_checks_on(json!({"enable_pre_call_checks": true}))]
+ fn unset_and_ignored_keys_are_accepted(#[case] json: Value) {
+ parse_router_config(json).unwrap();
+ }
+
+ #[rstest]
+ #[case::not_object(json!([]), false)]
+ #[case::bad_shape(json!({"retry_after": "soon"}), true)]
+ fn a_config_that_is_not_an_object_is_a_value_error_and_a_wrong_shape_declines(
+ #[case] json: Value,
+ #[case] declined: bool,
+ ) {
+ assert_eq!(parse_router_config(json).unwrap_err().declined(), declined);
+ }
+
+ #[rstest]
+ fn python_router_defaults_read_as_unset() {
+ let dynamic = parse_dynamic_config(json!({
+ "num_retries": null, "fallbacks": [], "context_window_fallbacks": [],
+ "content_policy_fallbacks": [], "model_group_alias": {"gpt": "gpt-4"},
+ }))
+ .unwrap();
+ assert_eq!(dynamic.num_retries, None);
+ assert!(dynamic.context_window_fallbacks.is_none());
+ assert!(dynamic.content_policy_fallbacks.is_none());
+ assert_eq!(dynamic.model_alias_map["gpt"], "gpt-4");
+ }
+
+ #[rstest]
+ #[case::own_model(json!({"model": "gpt", "num_retries": 1}), None, "gpt")]
+ #[case::override_model(json!({"model": "gpt"}), Some("other"), "other")]
+ fn a_request_reads_its_model(
+ #[case] json: Value,
+ #[case] model: Option<&str>,
+ #[case] expected: &str,
+ ) {
+ assert_eq!(parse_request(json, model).unwrap().model, expected);
+ }
+
+ #[rstest]
+ #[case::negative("negative", UncountedRetries::Negative)]
+ #[case::not_positive("not_positive", UncountedRetries::NotPositive)]
+ #[case::raises("raises", UncountedRetries::Raises)]
+ fn a_request_reads_a_num_retries_that_is_no_count(
+ #[case] spelled: &str,
+ #[case] expected: UncountedRetries,
+ ) {
+ let request =
+ parse_request(json!({"model": "gpt", "uncounted_retries": spelled}), None).unwrap();
+ assert_eq!(request.uncounted_retries, Some(expected));
+ assert_eq!(request.num_retries, None);
+ }
+
+ #[rstest]
+ #[case::unknown_field(json!({"model": "gpt", "temperature": 1}))]
+ #[case::no_model(json!({"team_id": "t"}))]
+ fn a_bad_request_is_a_value_error(#[case] json: Value) {
+ assert!(!parse_request(json, None).unwrap_err().declined());
+ }
+
+ #[rstest]
+ fn a_change_set_reads_upserts_and_deletes() {
+ let changes = parse_changes(json!({
+ "upsert": [{"model_name": "m", "litellm_params": {"model": "openai/x"}}],
+ "delete": ["old"],
+ }))
+ .unwrap();
+ assert_eq!(changes.upserts.len(), 1);
+ assert_eq!(changes.deletes, ["old"]);
+ }
+}
diff --git a/litellm-rust/crates/python-bridge/src/router/host.rs b/litellm-rust/crates/python-bridge/src/router/host.rs
new file mode 100644
index 00000000000..6ccac4df659
--- /dev/null
+++ b/litellm-rust/crates/python-bridge/src/router/host.rs
@@ -0,0 +1,1657 @@
+//! The host behind `RouterCoreHandle`. Rust never awaits Python: each host operation the call
+//! loop needs becomes a [`HostCall`] parked in the call's [`Exchange`], and the call future stays
+//! pending until Python runs it and hands the [`HostCallResponse`] back through [`Driver::resume`].
+//!
+//! The driver polls the call future with a no-op waker while the call's exchange is bound to the
+//! current thread, so a host future finds its exchange without the router passing it through.
+//!
+//! The routing cache is a host operation too: [`StepCache`] parks every [`RouterCache`] op as a
+//! `cache` step, and Python runs it against the router's own `DualCache` (memory plus Redis), so
+//! Python and Rust read and write the same cooldowns, fail counters and usage.
+
+use std::{
+ cell::RefCell,
+ future::{Future, poll_fn},
+ pin::Pin,
+ sync::{Arc, Mutex, MutexGuard, PoisonError},
+ task::{Context, Poll, Waker},
+ time::Duration,
+};
+
+use async_trait::async_trait;
+use litellm_router_core::{
+ Attempt, AttemptError, Blocked, BuildAsk, BuildPath, Built, Deployment, Error, Host, Output,
+ Payload, PreRoute, PreRouted, Provided, Randomness, RouterCache, RouterEvent, RouterRequest,
+ RouterResponse, RoutingContext, SecretRef, SelectStage, ThreadRandomness,
+ build::BuiltDeployment, host::ModelInfoLookup, model_info::MergedModelInfo,
+ precall::PreCallFacts,
+};
+use pyo3::{PyResult, Python, types::PyAnyMethods};
+use serde_json::{Map, Value};
+
+/// The `kind` of a step dict.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::Display, strum::IntoStaticStr)]
+pub(crate) enum StepKind {
+ #[strum(serialize = "attempt")]
+ Attempt,
+ #[strum(serialize = "sleep")]
+ Sleep,
+ #[strum(serialize = "allow_fallback")]
+ AllowFallback,
+ #[strum(serialize = "refuse_fallbacks")]
+ RefuseFallbacks,
+ #[strum(serialize = "select")]
+ Select,
+ #[strum(serialize = "pre_route")]
+ PreRoute,
+ #[strum(serialize = "cache")]
+ Cache,
+ #[strum(serialize = "secrets")]
+ Secrets,
+ #[strum(serialize = "validate")]
+ Validate,
+ #[strum(serialize = "provider")]
+ Provider,
+ #[strum(serialize = "added")]
+ Added,
+ #[strum(serialize = "model_info")]
+ ModelInfo,
+ #[strum(serialize = "done")]
+ Done,
+ #[strum(serialize = "failed")]
+ Failed,
+}
+
+/// Which selection stage Python is asked to run.
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) enum Stage {
+ /// Pre-call checks and sub-routers: return the deployments that remain.
+ Plugins,
+ /// A strategy router-core does not implement: return the candidates with the pick first.
+ /// When candidates share an id, `records` holds each candidate's deployment, in order, and the
+ /// host answers with the pick's position, so a pick splits over every entry as Python's does.
+ Strategy {
+ strategy: String,
+ records: Vec,
+ },
+ /// Tag routing: return the deployments that remain.
+ Tags,
+ /// `enable_pre_call_checks`: return the facts the filter decides by.
+ PreCall,
+ /// `_deployment_declares_no_vision`: return the candidates that declare no vision support.
+ TextOnly,
+ /// `_deployment_serves_route`: return the candidates that cannot serve `handler`'s call.
+ RouteIncapable { handler: String },
+ /// `_decisions_deployment_refuses_media`: return the candidates that refuse the call's media.
+ MediaRefusing,
+ /// `deployments_admitting_embedding_batch`: return the candidates the embedding pick keeps.
+ EmbeddingBatch,
+ /// The override strategy's pre-call check on a deployment named by id.
+ OverrideCheck(String),
+ /// The routing plugins: run them when `run`, then return the candidate models they left.
+ RoutingPlugins { run: bool },
+ /// The host's `CustomLogger` deployment filters: return the ids that remain, or a failure.
+ /// They read the ids this request already failed on, as Python's request kwargs carry them.
+ CallbackFilters {
+ excluded_ids: Vec,
+ skipped_ids: Vec,
+ target_order: Option,
+ },
+ /// `f"{provider}/{model}"` for a name no wildcard pattern matches as spelled: return it as
+ /// the one entry of `models`.
+ ProviderRequest,
+}
+
+/// The `op` of a `cache` step: one per [`RouterCache`] method.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::Display, strum::IntoStaticStr)]
+pub(crate) enum CacheOp {
+ /// `async_get_cache(key)`: answered with the stored JSON value, or `None`.
+ #[strum(serialize = "get")]
+ Get,
+ /// `async_set_cache(key, value, ttl)`: answered with `None`.
+ #[strum(serialize = "set")]
+ Set,
+ /// `async_increment(key, value, ttl)`: answered with the new number. The TTL applies when
+ /// the key is created.
+ #[strum(serialize = "increment")]
+ Increment,
+ /// `async_increment_usage(keys, value, ttl)`: `value` is one amount per key, added as
+ /// `_increment_deployment_usage` adds them. Answered with `None`.
+ #[strum(serialize = "increment_usage")]
+ IncrementUsage,
+ /// `async_batch_get_cache(keys)`: answered with one JSON value or `None` per key, in order.
+ #[strum(serialize = "batch_get")]
+ BatchGet,
+ /// `async_delete_cache(key)`: answered with `None`.
+ #[strum(serialize = "delete")]
+ Delete,
+}
+
+/// Python's outcome of a `cache` step.
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) enum CacheOutcome {
+ /// `get`: the stored value (`None` when missing).
+ Value(Option),
+ /// `batch_get`: one entry per key.
+ Values(Vec>),
+ /// `increment`: the counter after the increment.
+ Number(f64),
+ /// `set`, `delete` and `increment_usage`.
+ Stored,
+}
+
+impl CacheOutcome {
+ /// The op this outcome answers (`Stored` answers both `set` and `delete`).
+ fn answers(&self, op: CacheOp) -> bool {
+ matches!(
+ (self, op),
+ (Self::Value(_), CacheOp::Get)
+ | (Self::Values(_), CacheOp::BatchGet)
+ | (Self::Number(_), CacheOp::Increment)
+ | (
+ Self::Stored,
+ CacheOp::Set | CacheOp::Delete | CacheOp::IncrementUsage
+ )
+ )
+ }
+}
+
+/// A fallback chain's dict targets, which can carry an api key.
+#[derive(Clone, PartialEq, veil::Redact)]
+pub(crate) struct DictTargets(#[redact(with = "[REDACTED]")] pub(crate) Vec>);
+
+/// What a call is blocked on, for Python to run.
+#[derive(Clone, Debug, PartialEq)]
+pub(crate) enum HostCall {
+ Attempt {
+ deployment_id: String,
+ /// The picked deployment as configured: `model_name`, `litellm_params`, `model_info`
+ /// (with its `id`). The host builds the request from it; it never looks the deployment up
+ /// in a Python router.
+ deployment: Value,
+ /// The name the attempt resolved: the requested name or the fallback target.
+ model: String,
+ model_group: String,
+ /// The group a pre-routing hook selected for `model`, when it differs.
+ routed: Option,
+ /// The `order` level of a same-group order fallback.
+ target_order: Option,
+ /// Every deployment `model` resolved to for this caller, the picked one included.
+ candidate_ids: Vec,
+ retry: u32,
+ fallback_depth: u32,
+ fallback_level: u32,
+ timeout: Option,
+ /// Request params the fallback walk's dict targets merged in.
+ params: Option>,
+ /// The walk level whose request metadata the attempt carries.
+ scope: u32,
+ max_retries: u32,
+ model_group_size: usize,
+ },
+ Sleep {
+ seconds: f64,
+ },
+ AllowFallback {
+ target: String,
+ },
+ RefuseFallbacks {
+ /// The walk level whose chain the targets are.
+ level: u32,
+ targets: DictTargets,
+ },
+ Select {
+ stage: Stage,
+ model_group: String,
+ /// The name the request resolved from, which strategies key their state by.
+ name: String,
+ /// Whose selector a strategy reads: `default`, `group` (with `group`) or `override`.
+ scope: &'static str,
+ group: Option,
+ /// The deployments the name resolved to, before health filters (the tag stage reads them).
+ pool: Vec,
+ candidates: Vec,
+ },
+ /// Pre-routing before a pick resolves `model`: answered with the model group to resolve
+ /// instead (a string), a custom strategy's deployment document (a mapping), or `None`.
+ PreRoute {
+ model: String,
+ fallback_depth: u32,
+ retry: u32,
+ },
+ /// One [`RouterCache`] op against the host's cache. `key` is set for every op but
+ /// `batch_get`, which sets `keys`; `value` is set for `set` (the JSON to store) and
+ /// `increment` (the amount); `ttl` is in seconds.
+ Cache {
+ op: CacheOp,
+ key: Option,
+ keys: Option>,
+ value: Option,
+ ttl: Option,
+ /// The op reads or writes this worker's in-memory tier only (`local_only=True`).
+ local: bool,
+ },
+ /// The deployment build's `os.environ/` references: answered with one value per ref.
+ Secrets {
+ index: usize,
+ refs: Vec,
+ },
+ /// The deployment build's `validate`, `provider` and `added` steps for one deployment.
+ Build {
+ kind: StepKind,
+ path: BuildPath,
+ index: usize,
+ entry: Value,
+ declared_id: Option,
+ duplicate_id: bool,
+ /// The provider the `provider` step resolved (`added` only).
+ provider: Option,
+ },
+ /// One lookup of a model-info query ([`ModelInfoLookup`], as its `lookup` field and
+ /// arguments): answered as the lookup documents.
+ ModelInfo {
+ lookup: Value,
+ },
+}
+
+impl HostCall {
+ pub(crate) fn kind(&self) -> StepKind {
+ match self {
+ Self::Attempt { .. } => StepKind::Attempt,
+ Self::Sleep { .. } => StepKind::Sleep,
+ Self::AllowFallback { .. } => StepKind::AllowFallback,
+ Self::RefuseFallbacks { .. } => StepKind::RefuseFallbacks,
+ Self::Select { .. } => StepKind::Select,
+ Self::PreRoute { .. } => StepKind::PreRoute,
+ Self::Cache { .. } => StepKind::Cache,
+ Self::Secrets { .. } => StepKind::Secrets,
+ Self::Build { kind, .. } => *kind,
+ Self::ModelInfo { .. } => StepKind::ModelInfo,
+ }
+ }
+}
+
+/// Python's response to a [`HostCall`].
+#[derive(Debug)]
+pub(crate) enum HostCallResponse {
+ /// The attempt's total tokens, or its failure.
+ Attempt(Result),
+ /// A successful attempt whose response the provider blocked, or a stream whose outcome the
+ /// host reports when it ends (`deferred`), with its tokens.
+ Served {
+ total_tokens: u64,
+ blocked: Option,
+ deferred: bool,
+ tracks_usage: Option,
+ },
+ Slept,
+ AllowFallback(bool),
+ /// Whether the chain a `refuse_fallbacks` step named is refused.
+ RefuseFallbacks(bool),
+ /// The candidate ids that remain, pick first.
+ Select(Vec),
+ /// A strategy's pick among candidates sharing an id: positions in the candidates, pick first.
+ Positions(Vec),
+ /// The model group to resolve instead, a custom strategy's deployment, or `None`.
+ PreRoute(Option),
+ /// The facts a `pre_call` select step asked for.
+ PreCall(PreCallFacts),
+ /// The candidate models a `routing_plugins` select step reported.
+ Models(Option>),
+ /// The ids a `callback_filters` select step left, or the error a filter raised.
+ Filtered(Result, AttemptError>),
+ /// What a pick's select or pre-route step, or a fallback check, raised on the host instead of
+ /// answering: the walk takes it as the try's failure.
+ Raised(AttemptError),
+ Cache(CacheOutcome),
+ /// One value per ref of a `secrets` step.
+ Secrets(Vec),
+ Validated(Built),
+ Provided(Provided),
+ Added(Built),
+ /// A `model_info` lookup's response, as the host spelled it.
+ ModelInfo(Value),
+}
+
+impl HostCallResponse {
+ fn kind(&self) -> StepKind {
+ match self {
+ Self::Attempt(_) | Self::Served { .. } => StepKind::Attempt,
+ // Answers whichever step raised it (`check`).
+ Self::Raised(_) => StepKind::Select,
+ Self::Slept => StepKind::Sleep,
+ Self::AllowFallback(_) => StepKind::AllowFallback,
+ Self::RefuseFallbacks(_) => StepKind::RefuseFallbacks,
+ Self::Select(_)
+ | Self::Positions(_)
+ | Self::PreCall(_)
+ | Self::Models(_)
+ | Self::Filtered(_) => StepKind::Select,
+ Self::PreRoute(_) => StepKind::PreRoute,
+ Self::Cache(_) => StepKind::Cache,
+ Self::Secrets(_) => StepKind::Secrets,
+ Self::Validated(_) => StepKind::Validate,
+ Self::Provided(_) => StepKind::Provider,
+ Self::Added(_) => StepKind::Added,
+ Self::ModelInfo(_) => StepKind::ModelInfo,
+ }
+ }
+
+ fn check(&self, call: &HostCall) -> Result<(), DriveError> {
+ if matches!(self, Self::Raised(_))
+ && matches!(
+ call.kind(),
+ StepKind::Select | StepKind::PreRoute | StepKind::AllowFallback
+ )
+ {
+ return Ok(());
+ }
+ if self.kind() != call.kind() {
+ return Err(DriveError::WrongAnswer {
+ expected: call.kind(),
+ got: self.kind(),
+ });
+ }
+ if let (Self::Select(ids) | Self::Filtered(Ok(ids)), HostCall::Select { candidates, .. }) =
+ (self, call)
+ && let Some(id) = ids.iter().find(|id| !candidates.contains(id))
+ {
+ return Err(DriveError::UnknownCandidate { id: id.clone() });
+ }
+ if let (Self::Positions(positions), HostCall::Select { candidates, .. }) = (self, call)
+ && let Some(position) = positions.iter().find(|p| **p >= candidates.len())
+ {
+ return Err(DriveError::UnknownCandidate {
+ id: format!("#{position}"),
+ });
+ }
+ if let (Self::Secrets(values), HostCall::Secrets { refs, .. }) = (self, call)
+ && values.len() != refs.len()
+ {
+ return Err(DriveError::SecretsLength {
+ expected: refs.len(),
+ got: values.len(),
+ });
+ }
+ if let (Self::Cache(outcome), HostCall::Cache { op, keys, .. }) = (self, call) {
+ if !outcome.answers(*op) {
+ return Err(DriveError::WrongCacheOutcome { op: *op });
+ }
+ if let (CacheOutcome::Values(values), Some(keys)) = (outcome, keys)
+ && values.len() != keys.len()
+ {
+ return Err(DriveError::BatchLength {
+ expected: keys.len(),
+ got: values.len(),
+ });
+ }
+ }
+ Ok(())
+ }
+}
+
+/// How a driven future ended well.
+#[derive(Debug)]
+pub(crate) enum Finish {
+ /// `start`: the call's response.
+ Call(RouterResponse),
+ /// `select`: the deployment picked and the model group it serves.
+ Selected {
+ deployment_id: String,
+ model_group: String,
+ /// The deployment as served: a priority group's member carries its priority as `order`.
+ deployment: Value,
+ },
+ /// `healthy`: the deployments a pick would choose among, as served.
+ Healthy {
+ /// A deployment named by its id, which the listing returns alone rather than as a list.
+ single: bool,
+ deployment_ids: Vec,
+ deployments: Vec,
+ },
+ /// `report_failure`: the cooldown applied, if the deployment was cooled down.
+ Reported { cooldown: Option },
+ /// `report_success` / `count_success`: the success counter's key, if one was counted.
+ Counted { key: Option },
+ /// `build`: the deployments it installed, in order.
+ Built(Vec),
+ /// `router_model_info`: the merged model info (`None` when the lookup answered `None`).
+ RouterModelInfo(Option),
+ /// `deployment_model_info`: the merged model info, if any.
+ DeploymentModelInfo(Option>),
+ /// A model-info query that ends with one JSON answer (`model_group_info`,
+ /// `configured_token_limits`, `model_listing_info`).
+ Value(Value),
+}
+
+/// One step of a call: what it is blocked on, or how it ended.
+#[derive(Debug)]
+pub(crate) enum Step {
+ HostCall(HostCall),
+ Done(Finish),
+ Failed(Error),
+}
+
+/// A misuse of the step protocol by the caller.
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum DriveError {
+ #[error("the router call already finished")]
+ Finished,
+ #[error("the router call has not made a host call yet, so the outcome must be None")]
+ UnexpectedResponse,
+ #[error("the router call is waiting for an outcome of its `{expected}` step")]
+ MissingResponse { expected: StepKind },
+ #[error("the router call is waiting for an outcome of its `{expected}` step, not `{got}`")]
+ WrongAnswer { expected: StepKind, got: StepKind },
+ #[error("deployment `{id}` was not one of the select step's candidates")]
+ UnknownCandidate { id: String },
+ #[error("the outcome does not answer a `{op}` cache step")]
+ WrongCacheOutcome { op: CacheOp },
+ #[error("a `batch_get` cache step for {expected} keys was answered with {got} values")]
+ BatchLength { expected: usize, got: usize },
+ #[error("a `secrets` step for {expected} references was answered with {got} values")]
+ SecretsLength { expected: usize, got: usize },
+ #[error("the router call is pending on something other than a step")]
+ Stalled,
+}
+
+impl DriveError {
+ /// Whether the caller passed a bad outcome, rather than driving a call that cannot go on.
+ pub(crate) fn is_request(&self) -> bool {
+ match self {
+ Self::UnexpectedResponse
+ | Self::MissingResponse { .. }
+ | Self::WrongAnswer { .. }
+ | Self::UnknownCandidate { .. }
+ | Self::WrongCacheOutcome { .. }
+ | Self::BatchLength { .. }
+ | Self::SecretsLength { .. } => true,
+ Self::Finished | Self::Stalled => false,
+ }
+ }
+}
+
+/// The host call a router call is blocked on and the response that releases it.
+#[derive(Default)]
+struct Exchange {
+ call: Option,
+ response: Option,
+ /// Events the call reported since the last step, handed to Python with the next one.
+ events: Vec,
+}
+
+type Shared = Arc>;
+
+fn lock(exchange: &Mutex) -> MutexGuard<'_, Exchange> {
+ exchange.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+thread_local! {
+ static CURRENT: RefCell> = const { RefCell::new(None) };
+}
+
+/// Binds an exchange to this thread for one poll and restores the previous one after.
+struct Bound(Option);
+
+impl Bound {
+ fn new(exchange: &Shared) -> Self {
+ Self(CURRENT.replace(Some(exchange.clone())))
+ }
+}
+
+impl Drop for Bound {
+ fn drop(&mut self) {
+ CURRENT.set(self.0.take());
+ }
+}
+
+/// Park the host call `call` in the bound exchange and wait for its response.
+/// `response`, or the failure a step answered with instead ([`HostCallResponse::Raised`]).
+fn answered(response: HostCallResponse) -> Result {
+ match response {
+ HostCallResponse::Raised(failure) => Err(failure),
+ response => Ok(response),
+ }
+}
+
+async fn call_host(call: HostCall) -> HostCallResponse {
+ let mut call = Some(call);
+ poll_fn(move |_| {
+ CURRENT.with_borrow(|current| {
+ let Some(exchange) = current else {
+ return Poll::Pending;
+ };
+ let mut exchange = lock(exchange);
+ if let Some(call) = call.take() {
+ exchange.call = Some(call);
+ return Poll::Pending;
+ }
+ exchange.response.take().map_or(Poll::Pending, Poll::Ready)
+ })
+ })
+ .await
+}
+
+const CHECKED: &str = "the driver checks each response against its host call";
+
+fn ttl_seconds(ttl: Option) -> Option {
+ ttl.map(|ttl| ttl.as_secs_f64())
+}
+
+/// An increment amount as Python's counters store it: an int when it is whole.
+fn amount_value(amount: f64) -> Value {
+ if amount.fract() == 0.0 && amount.abs() < i64::MAX as f64 {
+ Value::from(amount as i64)
+ } else {
+ Value::from(amount)
+ }
+}
+
+/// The host's cache: every op is a `cache` step Python runs against the router's `DualCache`.
+pub(crate) struct StepCache;
+
+async fn cache_step(
+ op: CacheOp,
+ key: Option<&str>,
+ keys: Option<&[String]>,
+ value: Option,
+ ttl: Option,
+) -> CacheOutcome {
+ call_cache(op, key, keys, value, ttl, false).await
+}
+
+async fn call_cache(
+ op: CacheOp,
+ key: Option<&str>,
+ keys: Option<&[String]>,
+ value: Option,
+ ttl: Option,
+ local: bool,
+) -> CacheOutcome {
+ let call = HostCall::Cache {
+ op,
+ key: key.map(str::to_owned),
+ keys: keys.map(<[String]>::to_vec),
+ value,
+ ttl: ttl_seconds(ttl),
+ local,
+ };
+ let HostCallResponse::Cache(outcome) = call_host(call).await else {
+ unreachable!("{CHECKED}");
+ };
+ outcome
+}
+
+#[async_trait]
+impl RouterCache for StepCache {
+ async fn async_get_cache(&self, key: &str) -> Result, litellm_cache::Error> {
+ let CacheOutcome::Value(value) =
+ cache_step(CacheOp::Get, Some(key), None, None, None).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(value)
+ }
+
+ async fn async_set_cache(
+ &self,
+ key: &str,
+ value: Value,
+ ttl: Option,
+ ) -> Result<(), litellm_cache::Error> {
+ cache_step(CacheOp::Set, Some(key), None, Some(value), ttl).await;
+ Ok(())
+ }
+
+ async fn async_increment(
+ &self,
+ key: &str,
+ amount: f64,
+ ttl: Option,
+ ) -> Result {
+ let value = Some(amount_value(amount));
+ let CacheOutcome::Number(next) =
+ cache_step(CacheOp::Increment, Some(key), None, value, ttl).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(next)
+ }
+
+ async fn async_batch_get_cache(
+ &self,
+ keys: &[String],
+ ) -> Result>, litellm_cache::Error> {
+ let CacheOutcome::Values(values) =
+ cache_step(CacheOp::BatchGet, None, Some(keys), None, None).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(values)
+ }
+
+ async fn async_delete_cache(&self, key: &str) -> Result<(), litellm_cache::Error> {
+ cache_step(CacheOp::Delete, Some(key), None, None, None).await;
+ Ok(())
+ }
+
+ async fn async_increment_usage(
+ &self,
+ increments: &[(String, f64)],
+ ttl: Option,
+ ) -> Result<(), litellm_cache::Error> {
+ let keys: Vec = increments.iter().map(|(key, _)| key.clone()).collect();
+ let amounts = increments
+ .iter()
+ .map(|(_, amount)| amount_value(*amount))
+ .collect();
+ let value = Some(Value::Array(amounts));
+ cache_step(CacheOp::IncrementUsage, None, Some(&keys), value, ttl).await;
+ Ok(())
+ }
+
+ async fn async_get_local(&self, key: &str) -> Result, litellm_cache::Error> {
+ let CacheOutcome::Value(value) =
+ call_cache(CacheOp::Get, Some(key), None, None, None, true).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(value)
+ }
+
+ async fn async_increment_local(
+ &self,
+ key: &str,
+ amount: f64,
+ ttl: Option,
+ ) -> Result {
+ let value = Some(amount_value(amount));
+ let CacheOutcome::Number(next) =
+ call_cache(CacheOp::Increment, Some(key), None, value, ttl, true).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(next)
+ }
+}
+
+/// A deployment as Python spells a `model_list` entry, with its id in `model_info`.
+/// Whether two candidates share an id, which a pick answered by id can't tell apart.
+fn shares_ids(candidates: &[Arc]) -> bool {
+ let mut seen = std::collections::HashSet::new();
+ !candidates.iter().all(|d| seen.insert(d.id.as_str()))
+}
+
+/// The candidates a host answer names, in its order. An id names every candidate that carries
+/// it, so deployments sharing an id stay together through a filter, as Python's list keeps them.
+fn answered_by_id(candidates: &[Arc], ids: &[String]) -> Vec> {
+ let mut taken = vec![false; candidates.len()];
+ let mut kept = Vec::new();
+ for id in ids {
+ for (position, deployment) in candidates.iter().enumerate() {
+ if !taken[position] && deployment.id == *id {
+ taken[position] = true;
+ kept.push(deployment.clone());
+ }
+ }
+ }
+ kept
+}
+
+pub(crate) fn deployment_record(deployment: &Deployment) -> Value {
+ deployment.record()
+}
+
+/// The process's `random` module: a Python router draws its shuffle and retry jitter there, so
+/// a seeded process picks the same deployments on either core. The bare module functions share
+/// one generator, so these calls interleave with every other draw the process makes.
+pub(crate) struct PythonRandomness;
+
+impl Randomness for PythonRandomness {
+ fn next_f64(&self) -> f64 {
+ Python::attach(|py| -> PyResult {
+ py.import("random")?.call_method0("random")?.extract()
+ })
+ .unwrap_or_else(|_| ThreadRandomness.next_f64())
+ }
+
+ fn index_below(&self, len: usize) -> usize {
+ // `random.choice` is `seq[self._randbelow(len(seq))]`, and so is `randrange(len)`.
+ Python::attach(|py| -> PyResult {
+ py.import("random")?
+ .call_method1("randrange", (len,))?
+ .extract()
+ })
+ .unwrap_or_else(|_| ThreadRandomness.index_below(len))
+ }
+}
+
+/// The [`Host`] every `RouterCoreHandle` router runs on.
+pub(crate) struct StepHost;
+
+#[async_trait]
+impl Host for StepHost {
+ async fn invoke(&self, attempt: &Attempt) -> Result {
+ let call = HostCall::Attempt {
+ deployment_id: attempt.deployment.id.clone(),
+ deployment: deployment_record(&attempt.deployment),
+ model: attempt.name.clone(),
+ model_group: attempt.model_group.clone(),
+ routed: attempt.routed.clone(),
+ target_order: attempt.target_order,
+ candidate_ids: attempt.candidate_ids.clone(),
+ retry: attempt.retry,
+ fallback_depth: attempt.fallback_depth,
+ fallback_level: attempt.fallback_level,
+ timeout: attempt.timeout.map(|timeout| timeout.as_secs_f64()),
+ params: attempt
+ .params
+ .as_ref()
+ .map(|params| Box::new(Value::Object(Map::clone(params)))),
+ scope: attempt.scope,
+ max_retries: attempt.max_retries,
+ model_group_size: attempt.model_group_size,
+ };
+ let (result, blocked, deferred, tracks_usage) = match call_host(call).await {
+ HostCallResponse::Attempt(result) => (result, None, false, None),
+ HostCallResponse::Served {
+ total_tokens,
+ blocked,
+ deferred,
+ tracks_usage,
+ } => (Ok(total_tokens), blocked, deferred, tracks_usage),
+ _ => unreachable!("{CHECKED}"),
+ };
+ result.map(|total_tokens| Output {
+ value: Payload(Arc::new(())),
+ total_tokens,
+ blocked,
+ deferred,
+ tracks_usage,
+ })
+ }
+
+ async fn select(
+ &self,
+ stage: &SelectStage,
+ ctx: &RoutingContext<'_>,
+ candidates: Vec>,
+ ) -> Result>, Error> {
+ let (scope, group) = ctx.scope.wire();
+ let call = HostCall::Select {
+ stage: match stage {
+ SelectStage::Plugins => Stage::Plugins,
+ SelectStage::Strategy(strategy) => Stage::Strategy {
+ strategy: strategy.to_string(),
+ records: if shares_ids(&candidates) {
+ candidates.iter().map(|d| deployment_record(d)).collect()
+ } else {
+ Vec::new()
+ },
+ },
+ SelectStage::Tags => Stage::Tags,
+ SelectStage::TextOnly => Stage::TextOnly,
+ SelectStage::RouteIncapable { handler } => Stage::RouteIncapable {
+ handler: handler.clone(),
+ },
+ SelectStage::MediaRefusing => Stage::MediaRefusing,
+ SelectStage::EmbeddingBatch => Stage::EmbeddingBatch,
+ SelectStage::OverrideCheck(strategy) => Stage::OverrideCheck(strategy.to_string()),
+ },
+ pool: ctx.pool.iter().map(|d| d.id.clone()).collect(),
+ model_group: ctx.model_group.to_owned(),
+ name: ctx.name.to_owned(),
+ scope,
+ group: group.map(str::to_owned),
+ candidates: candidates.iter().map(|d| d.id.clone()).collect(),
+ };
+ match answered(call_host(call).await)? {
+ HostCallResponse::Select(ids) => Ok(answered_by_id(&candidates, &ids)),
+ HostCallResponse::Positions(positions) => Ok(positions
+ .iter()
+ .map(|position| candidates[*position].clone())
+ .collect()),
+ _ => unreachable!("{CHECKED}"),
+ }
+ }
+
+ async fn pre_route(&self, pre: &PreRoute<'_>) -> Result, Error> {
+ let call = HostCall::PreRoute {
+ model: pre.name.to_owned(),
+ fallback_depth: pre.fallback_depth,
+ retry: pre.retry,
+ };
+ let HostCallResponse::PreRoute(routed) = answered(call_host(call).await)? else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(routed)
+ }
+
+ async fn routing_plugin_models(
+ &self,
+ ctx: &RoutingContext<'_>,
+ run: bool,
+ ) -> Result>, Error> {
+ let (scope, group) = ctx.scope.wire();
+ let call = HostCall::Select {
+ stage: Stage::RoutingPlugins { run },
+ pool: Vec::new(),
+ model_group: ctx.model_group.to_owned(),
+ name: ctx.name.to_owned(),
+ scope,
+ group: group.map(str::to_owned),
+ candidates: Vec::new(),
+ };
+ let HostCallResponse::Models(models) = answered(call_host(call).await)? else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(models)
+ }
+
+ async fn provider_request(&self, ctx: &RoutingContext<'_>) -> Result, Error> {
+ let (scope, group) = ctx.scope.wire();
+ let call = HostCall::Select {
+ stage: Stage::ProviderRequest,
+ pool: Vec::new(),
+ model_group: ctx.model_group.to_owned(),
+ name: ctx.name.to_owned(),
+ scope,
+ group: group.map(str::to_owned),
+ candidates: Vec::new(),
+ };
+ let HostCallResponse::Models(models) = answered(call_host(call).await)? else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(models.into_iter().flatten().next())
+ }
+
+ async fn pre_call_facts(
+ &self,
+ ctx: &RoutingContext<'_>,
+ candidates: &[Arc],
+ ) -> Result {
+ let (scope, group) = ctx.scope.wire();
+ let call = HostCall::Select {
+ stage: Stage::PreCall,
+ pool: ctx.pool.iter().map(|d| d.id.clone()).collect(),
+ model_group: ctx.model_group.to_owned(),
+ name: ctx.name.to_owned(),
+ scope,
+ group: group.map(str::to_owned),
+ candidates: candidates.iter().map(|d| d.id.clone()).collect(),
+ };
+ let HostCallResponse::PreCall(facts) = answered(call_host(call).await)? else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(facts)
+ }
+
+ async fn callback_filters(
+ &self,
+ ctx: &RoutingContext<'_>,
+ candidates: Vec>,
+ ) -> Result>, Error> {
+ let (scope, group) = ctx.scope.wire();
+ let joined = |own: &[String], walked: &[String]| [own, walked].concat();
+ let call = HostCall::Select {
+ stage: Stage::CallbackFilters {
+ excluded_ids: joined(&ctx.request.excluded_ids, ctx.excluded_ids),
+ skipped_ids: joined(&ctx.request.retry_skipped_ids, ctx.skipped_ids),
+ target_order: ctx.target_order,
+ },
+ pool: Vec::new(),
+ model_group: ctx.model_group.to_owned(),
+ name: ctx.name.to_owned(),
+ scope,
+ group: group.map(str::to_owned),
+ candidates: candidates.iter().map(|d| d.id.clone()).collect(),
+ };
+ let HostCallResponse::Filtered(result) = call_host(call).await else {
+ unreachable!("{CHECKED}");
+ };
+ let ids = result.map_err(Error::Attempt)?;
+ Ok(answered_by_id(&candidates, &ids))
+ }
+
+ async fn allow_fallback(
+ &self,
+ _request: &RouterRequest,
+ target: &str,
+ ) -> Result {
+ let call = HostCall::AllowFallback {
+ target: target.to_owned(),
+ };
+ let HostCallResponse::AllowFallback(allowed) = answered(call_host(call).await)? else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(allowed)
+ }
+
+ async fn refuses_fallbacks(&self, level: u32, targets: &[Map]) -> bool {
+ let call = HostCall::RefuseFallbacks {
+ level,
+ targets: DictTargets(targets.to_vec()),
+ };
+ let HostCallResponse::RefuseFallbacks(refused) = call_host(call).await else {
+ unreachable!("{CHECKED}");
+ };
+ refused
+ }
+
+ /// A zero sleep is asked too: Python's retry loop awaits `asyncio.sleep(0)` between sibling
+ /// retries, which yields to the event loop.
+ async fn sleep(&self, duration: Duration) {
+ let call = HostCall::Sleep {
+ seconds: duration.as_secs_f64(),
+ };
+ let HostCallResponse::Slept = call_host(call).await else {
+ unreachable!("{CHECKED}");
+ };
+ }
+
+ /// Kept for Python only when it acts on it: an exhausted retry loop is stamped on the
+ /// error Python raises, a fallback hop tells it which target a failed walk ended on, and a
+ /// cooldown or a pick is reported to the host's Prometheus and service loggers (a pick also
+ /// joins the router cache's per-deployment count, a failed walk the router's error log).
+ fn observe(&self, event: &RouterEvent) {
+ if !matches!(
+ event,
+ RouterEvent::RetriesExhausted { .. }
+ | RouterEvent::Retry { .. }
+ | RouterEvent::Fallback { .. }
+ | RouterEvent::FallingBack { .. }
+ | RouterEvent::OrderChainFailed { .. }
+ | RouterEvent::FallbacksFailed { .. }
+ | RouterEvent::LevelFailed { .. }
+ | RouterEvent::PickFailed { .. }
+ | RouterEvent::RetriesOpened { .. }
+ | RouterEvent::RetryBudget { .. }
+ | RouterEvent::MockFailed { .. }
+ | RouterEvent::RetryLogged { .. }
+ | RouterEvent::RetryPolicyRaised { .. }
+ | RouterEvent::RetryCountRaised { .. }
+ | RouterEvent::FallbackDone { .. }
+ | RouterEvent::Blocked { .. }
+ | RouterEvent::CooldownSet { .. }
+ | RouterEvent::Selected { .. }
+ | RouterEvent::PickStarted { .. }
+ ) {
+ return;
+ }
+ CURRENT.with_borrow(|current| {
+ if let Some(exchange) = current {
+ lock(exchange).events.push(event.clone());
+ }
+ });
+ }
+
+ fn cache(&self) -> Arc {
+ Arc::new(StepCache)
+ }
+
+ async fn secrets(&self, index: usize, refs: &[SecretRef]) -> Result, Error> {
+ let HostCallResponse::Secrets(values) = call_host(HostCall::Secrets {
+ index,
+ refs: refs.to_vec(),
+ })
+ .await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(values)
+ }
+
+ async fn validate(&self, build: &BuildAsk<'_>) -> Result {
+ let HostCallResponse::Validated(built) =
+ call_host(build_host_call(StepKind::Validate, build, None)).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(built)
+ }
+
+ async fn provider(&self, build: &BuildAsk<'_>) -> Result {
+ let HostCallResponse::Provided(provided) =
+ call_host(build_host_call(StepKind::Provider, build, None)).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(provided)
+ }
+
+ async fn added(&self, build: &BuildAsk<'_>, provider: Option<&str>) -> Result {
+ let HostCallResponse::Added(built) =
+ call_host(build_host_call(StepKind::Added, build, provider)).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(built)
+ }
+
+ async fn model_info(&self, lookup: &ModelInfoLookup) -> Result {
+ let lookup = serde_json::to_value(lookup).unwrap_or(Value::Null);
+ let HostCallResponse::ModelInfo(response) = call_host(HostCall::ModelInfo { lookup }).await
+ else {
+ unreachable!("{CHECKED}");
+ };
+ Ok(response)
+ }
+}
+
+fn build_host_call(kind: StepKind, build: &BuildAsk<'_>, provider: Option<&str>) -> HostCall {
+ HostCall::Build {
+ kind,
+ path: build.path,
+ index: build.index,
+ entry: Value::Object(build.entry.clone()),
+ declared_id: build.declared_id.map(str::to_owned),
+ duplicate_id: build.duplicate_id,
+ provider: provider.map(str::to_owned),
+ }
+}
+
+type CallFuture = Pin> + Send>>;
+
+/// Poll `future` once with `exchange` bound.
+fn poll_bound(future: Pin<&mut F>, exchange: &Shared) -> Poll {
+ let _bound = Bound::new(exchange);
+ future.poll(&mut Context::from_waker(Waker::noop()))
+}
+
+/// Drives one router call a step at a time.
+pub(crate) struct Driver {
+ future: Option,
+ exchange: Shared,
+ pending_call: Option,
+}
+
+impl Driver {
+ pub(crate) fn new(call: impl Future