diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260429161855_workflow_runs_tables/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260429161855_workflow_runs_tables/migration.sql new file mode 100644 index 00000000000..6454f267656 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260429161855_workflow_runs_tables/migration.sql @@ -0,0 +1,75 @@ +-- CreateTable +CREATE TABLE "LiteLLM_WorkflowRun" ( + "run_id" TEXT NOT NULL, + "session_id" TEXT NOT NULL, + "workflow_type" TEXT NOT NULL, + "status" TEXT NOT NULL DEFAULT 'pending', + "created_by" TEXT, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL, + "input" JSONB, + "output" JSONB, + "metadata" JSONB, + + CONSTRAINT "LiteLLM_WorkflowRun_pkey" PRIMARY KEY ("run_id") +); + +-- CreateTable +CREATE TABLE "LiteLLM_WorkflowEvent" ( + "event_id" TEXT NOT NULL, + "run_id" TEXT NOT NULL, + "event_type" TEXT NOT NULL, + "step_name" TEXT NOT NULL, + "sequence_number" INTEGER NOT NULL, + "data" JSONB, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "LiteLLM_WorkflowEvent_pkey" PRIMARY KEY ("event_id") +); + +-- CreateTable +CREATE TABLE "LiteLLM_WorkflowMessage" ( + "message_id" TEXT NOT NULL, + "run_id" TEXT NOT NULL, + "role" TEXT NOT NULL, + "content" TEXT NOT NULL, + "sequence_number" INTEGER NOT NULL, + "session_id" TEXT, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "LiteLLM_WorkflowMessage_pkey" PRIMARY KEY ("message_id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "LiteLLM_WorkflowRun_session_id_key" ON "LiteLLM_WorkflowRun"("session_id"); + +-- CreateIndex +CREATE INDEX "LiteLLM_WorkflowRun_workflow_type_status_idx" ON "LiteLLM_WorkflowRun"("workflow_type", "status"); + +-- CreateIndex +CREATE INDEX "LiteLLM_WorkflowRun_session_id_idx" ON "LiteLLM_WorkflowRun"("session_id"); + +-- CreateIndex +CREATE INDEX "LiteLLM_WorkflowRun_created_at_idx" ON "LiteLLM_WorkflowRun"("created_at"); + +-- CreateIndex +CREATE INDEX "LiteLLM_WorkflowRun_created_by_idx" ON "LiteLLM_WorkflowRun"("created_by"); + +-- CreateIndex +CREATE INDEX "LiteLLM_WorkflowEvent_run_id_idx" ON "LiteLLM_WorkflowEvent"("run_id"); + +-- CreateIndex +CREATE UNIQUE INDEX "LiteLLM_WorkflowEvent_run_id_sequence_number_key" ON "LiteLLM_WorkflowEvent"("run_id", "sequence_number"); + +-- CreateIndex +CREATE INDEX "LiteLLM_WorkflowMessage_run_id_idx" ON "LiteLLM_WorkflowMessage"("run_id"); + +-- CreateIndex +CREATE UNIQUE INDEX "LiteLLM_WorkflowMessage_run_id_sequence_number_key" ON "LiteLLM_WorkflowMessage"("run_id", "sequence_number"); + +-- AddForeignKey +ALTER TABLE "LiteLLM_WorkflowEvent" ADD CONSTRAINT "LiteLLM_WorkflowEvent_run_id_fkey" FOREIGN KEY ("run_id") REFERENCES "LiteLLM_WorkflowRun"("run_id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "LiteLLM_WorkflowMessage" ADD CONSTRAINT "LiteLLM_WorkflowMessage_run_id_fkey" FOREIGN KEY ("run_id") REFERENCES "LiteLLM_WorkflowRun"("run_id") ON DELETE RESTRICT ON UPDATE CASCADE; + diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 8f07c5afa3f..fd54ed5243a 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1290,3 +1290,80 @@ model LiteLLM_AdaptiveRouterSession { @@id([session_id, router_name, model_name]) @@index([last_activity_at], map: "idx_adaptive_router_session_activity") } + +// --------------------------------------------------------------------------- +// Workflow Run Tracking +// +// Generic durable state tracking for any agent or automated workflow. +// Design: three tables — run (header + materialized status), event (append-only +// source of truth for state transitions), message (conversation inbox/outbox). +// +// Usage: +// - Set `workflow_type` to identify the owning system (e.g. "shin-builder"). +// - Store domain-specific fields in `metadata` (worktree_path, pr_url, etc.). +// - `session_id` on WorkflowRun matches `x-litellm-session-id` header sent to +// the proxy — all spend logs for this run are automatically tagged. +// --------------------------------------------------------------------------- + +// One instance of work being done. `status` is a materialized cache of the +// latest event; the event log is the authoritative source of truth. +model LiteLLM_WorkflowRun { + run_id String @id @default(uuid()) + session_id String @unique @default(uuid()) + workflow_type String + status String @default("pending") + created_by String? // user_id of the key that created this run; null = created by master key + created_at DateTime @default(now()) + updated_at DateTime @updatedAt + input Json? + output Json? + metadata Json? + + events LiteLLM_WorkflowEvent[] + messages LiteLLM_WorkflowMessage[] + + @@index([workflow_type, status]) + @@index([session_id]) + @@index([created_at]) + @@index([created_by]) +} + +// Append-only log of state transitions. Never mutate rows here. +// `step_name` and `event_type` are caller-defined strings — no hardcoded enums. +// Status auto-update rules (applied by the append endpoint): +// step.started → run.status = running +// step.failed → run.status = failed +// hook.waiting → run.status = paused +// hook.received → run.status = running +model LiteLLM_WorkflowEvent { + event_id String @id @default(uuid()) + run_id String + event_type String + step_name String + sequence_number Int + data Json? + created_at DateTime @default(now()) + + run LiteLLM_WorkflowRun @relation(fields: [run_id], references: [run_id]) + + @@unique([run_id, sequence_number]) + @@index([run_id]) +} + +// Conversation inbox/outbox — full message content, separate from the durable +// event log. Spend logs truncate messages; this table stores them in full. +// `session_id` here is the Claude --resume session ID (or similar). +model LiteLLM_WorkflowMessage { + message_id String @id @default(uuid()) + run_id String + role String + content String + sequence_number Int + session_id String? + created_at DateTime @default(now()) + + run LiteLLM_WorkflowRun @relation(fields: [run_id], references: [run_id]) + + @@unique([run_id, sequence_number]) + @@index([run_id]) +} diff --git a/litellm/integrations/prompt_management_base.py b/litellm/integrations/prompt_management_base.py index 71da650dc48..9c626aea849 100644 --- a/litellm/integrations/prompt_management_base.py +++ b/litellm/integrations/prompt_management_base.py @@ -87,9 +87,7 @@ class PromptManagementBase(ABC): try: messages = compiled_prompt_client["prompt_template"] + client_messages except Exception as e: - raise ValueError( - f"Error compiling prompt: {e}. Prompt id={prompt_id}, prompt_variables={prompt_variables}, client_messages={client_messages}, dynamic_callback_params={dynamic_callback_params}" - ) + raise ValueError(f"Error compiling prompt: {e}. Prompt id={prompt_id}") compiled_prompt_client["completed_messages"] = messages return compiled_prompt_client @@ -116,9 +114,7 @@ class PromptManagementBase(ABC): try: messages = compiled_prompt_client["prompt_template"] + client_messages except Exception as e: - raise ValueError( - f"Error compiling prompt: {e}. Prompt id={prompt_id}, prompt_variables={prompt_variables}, client_messages={client_messages}, dynamic_callback_params={dynamic_callback_params}" - ) + raise ValueError(f"Error compiling prompt: {e}. Prompt id={prompt_id}") compiled_prompt_client["completed_messages"] = messages return compiled_prompt_client diff --git a/litellm/litellm_core_utils/get_llm_provider_logic.py b/litellm/litellm_core_utils/get_llm_provider_logic.py index 4ff077efe7c..c0ca6835eee 100644 --- a/litellm/litellm_core_utils/get_llm_provider_logic.py +++ b/litellm/litellm_core_utils/get_llm_provider_logic.py @@ -1,4 +1,5 @@ from typing import Optional, Tuple +from urllib.parse import urlparse import litellm from litellm.constants import REPLICATE_MODEL_NAME_WITH_ID_LENGTH @@ -8,6 +9,43 @@ from litellm.secret_managers.main import get_secret, get_secret_str from ..types.router import LiteLLM_Params +def _endpoint_matches_api_base(endpoint: str, api_base: str) -> bool: + """ + Match a registered openai-compatible endpoint against a caller-supplied + ``api_base`` using parsed-URL semantics, not unanchored substring search. + + Both inputs may be a bare hostname (``api.perplexity.ai``), host+path + (``api.deepinfra.com/v1/openai``), or a full URL + (``https://api.cerebras.ai/v1``). Hostnames must match exactly + (case-insensitive); if the registered endpoint has a non-trivial path, + the api_base path must start with it on a segment boundary. + + The naive ``endpoint in api_base`` shape lets a caller pass + ``https://attacker.com/api.groq.com/openai/v1`` to coerce the proxy + into reading the server's GROQ_API_KEY from the environment and + forwarding it to the attacker's host as a Bearer credential. + """ + + def _parse(value: str): + # Ensure urlparse sees a scheme so it populates hostname / path. + normalized = value if "://" in value else f"https://{value}" + return urlparse(normalized) + + parsed_endpoint = _parse(endpoint) + parsed_url = _parse(api_base) + + endpoint_host = (parsed_endpoint.hostname or "").lower() + url_host = (parsed_url.hostname or "").lower() + if not endpoint_host or endpoint_host != url_host: + return False + + endpoint_path = parsed_endpoint.path.rstrip("/") + if not endpoint_path: + return True + url_path = parsed_url.path.rstrip("/") + return url_path == endpoint_path or url_path.startswith(endpoint_path + "/") + + def _is_non_openai_azure_model(model: str) -> bool: try: model_name = model.split("/", 1)[1] @@ -210,7 +248,7 @@ def get_llm_provider( # noqa: PLR0915 # check if api base is a known openai compatible endpoint if api_base: for endpoint in litellm.openai_compatible_endpoints: - if endpoint in api_base: + if _endpoint_matches_api_base(endpoint, api_base): if endpoint == "api.perplexity.ai": custom_llm_provider = "perplexity" dynamic_api_key = get_secret_str("PERPLEXITYAI_API_KEY") diff --git a/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py b/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py index 78378faa262..5fd42fe0d36 100644 --- a/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py +++ b/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py @@ -824,8 +824,6 @@ def convert_to_model_response_object( # noqa: PLR0915 stream=stream, start_time=start_time, end_time=end_time, - hidden_params=hidden_params, - _response_headers=_response_headers, convert_tool_call_to_json_mode=convert_tool_call_to_json_mode, ) raise Exception( diff --git a/litellm/llms/vertex_ai/batches/handler.py b/litellm/llms/vertex_ai/batches/handler.py index 028e02eb0ca..7436bfef58b 100644 --- a/litellm/llms/vertex_ai/batches/handler.py +++ b/litellm/llms/vertex_ai/batches/handler.py @@ -4,6 +4,7 @@ from typing import Any, Coroutine, Dict, Optional, Union import httpx import litellm +from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get from litellm.llms.custom_httpx.http_handler import ( _get_httpx_client, get_async_httpx_client, @@ -224,8 +225,14 @@ class VertexAIBatchPrediction(VertexLLM): }, ) - response = sync_handler.get( - url=api_base, + # ``api_base`` here can come from caller-supplied request kwargs + # (clientside override). Wrap the fetch in ``safe_get`` so DNS + # rebind / private / cloud-metadata targets are rejected; the + # proxy auth gate already blocks malicious clientside ``api_base`` + # at the boundary — this is defense-in-depth for SDK callers. + response = safe_get( + sync_handler, + api_base, headers=headers, ) @@ -270,8 +277,13 @@ class VertexAIBatchPrediction(VertexLLM): }, ) - response = await client.get( - url=api_base, + # Mirror the sync path: ``api_base`` may come from caller-supplied + # request kwargs, so wrap the fetch in ``async_safe_get`` to reject + # DNS-rebind / private / cloud-metadata targets. Defense-in-depth + # behind the proxy auth gate's clientside ``api_base`` check. + response = await async_safe_get( + client, + api_base, headers=headers, ) if response.status_code != 200: diff --git a/litellm/proxy/auth/auth_utils.py b/litellm/proxy/auth/auth_utils.py index 448c975d123..91c8f2dd7c9 100644 --- a/litellm/proxy/auth/auth_utils.py +++ b/litellm/proxy/auth/auth_utils.py @@ -6,9 +6,11 @@ from typing import Any, List, Optional, Tuple from fastapi import HTTPException, Request, status +import litellm from litellm import Router, provider_list from litellm._logging import verbose_proxy_logger from litellm.constants import STANDARD_CUSTOMER_ID_HEADERS +from litellm.litellm_core_utils.url_utils import SSRFError, validate_url from litellm.proxy._types import * from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS @@ -53,6 +55,12 @@ def _check_valid_ip( def check_complete_credentials(request_body: dict) -> bool: """ if 'api_base' in request body. Check if complete credentials given. Prevent malicious attacks. + + Supplying an ``api_key`` is necessary but not sufficient: even with + credentials supplied, an ``api_base`` / ``base_url`` that resolves to a + private/internal/cloud-metadata address would still allow the proxy to + be used as an SSRF pivot. Validate any URL fields here so the gate + can't be bypassed with ``api_key=anything`` plus a malicious target. """ given_model: Optional[str] = None @@ -70,10 +78,27 @@ def check_complete_credentials(request_body: dict) -> bool: return False api_key_value = request_body.get("api_key") - if api_key_value and isinstance(api_key_value, str) and api_key_value.strip(): - return True + if not (api_key_value and isinstance(api_key_value, str) and api_key_value.strip()): + return False - return False + # ``validate_url`` itself doesn't consult the toggle; ``safe_get`` / + # ``async_safe_get`` do. Mirror that here so admins who explicitly + # disabled URL validation (e.g. for an internal Ollama endpoint they + # accept the SSRF risk for) aren't blocked at the proxy boundary. + if getattr(litellm, "user_url_validation", False): + for url_field in ("api_base", "base_url"): + url_value = request_body.get(url_field) + if not url_value or not isinstance(url_value, str): + continue + try: + validate_url(url_value) + except SSRFError as e: + raise ValueError( + f"Rejected request: client-side {url_field}={url_value!r} " + f"is rejected by the SSRF guard ({e})." + ) + + return True def check_regex_or_str_match(request_body_value: Any, regex_str: str) -> bool: @@ -159,15 +184,42 @@ def is_request_body_safe( "aws_web_identity_token", "aws_role_name", "vertex_credentials", + # Endpoint-targeting fields that retarget the outbound request or + # an observability callback. An attacker-controlled value either + # exfiltrates the request payload (incl. messages + admin-set + # tokens) to the attacker's host, or coerces the proxy into + # authenticating against the attacker's host with admin secrets. + "aws_bedrock_runtime_endpoint", + "langsmith_base_url", + "langfuse_host", + "posthog_host", + "braintrust_host", + "slack_webhook_url", + # Provider-specific endpoint overrides that flow into the outbound + # request via ``optional_params``. Same threat as ``api_base``: + # ``s3_endpoint_url`` redirects Bedrock file uploads to attacker + # S3; ``sagemaker_base_url`` redirects all SageMaker traffic; + # ``deployment_url`` redirects SAP deployments. + "s3_endpoint_url", + "sagemaker_base_url", + "deployment_url", ] + # The blocklist is enforced unconditionally. Legitimate clientside + # credential / endpoint passthrough goes through one of the two + # explicit admin opt-ins (``general_settings.allow_client_side_credentials`` + # proxy-wide or ``configurable_clientside_auth_params`` per deployment). + # Historically there was a third, *implicit*, *caller-controlled* path: + # ``check_complete_credentials`` returned True when the caller supplied + # any non-empty ``api_key``, which made the entire blocklist a no-op. + # That bypass turned every missing entry on the blocklist into an + # exploitable SSRF / credential-exfil hole — see GHSA-jh89-88fc-qrfp, + # GHSA-3frq-6r6h-7j64, and the chain of veria-admin findings (Dv_m860l, + # b_yRJeQ5, stN90yjP, LBlyOAc8, U2TD78kg). Removed: the blocklist now + # has a single, predictable failure mode for missing entries (a 400), + # not a credential leak. for param in banned_params: - if ( - param in request_body - and not check_complete_credentials( # allow client-credentials to be passed to proxy - request_body=request_body - ) - ): + if param in request_body: if general_settings.get("allow_client_side_credentials") is True: return True elif ( @@ -182,7 +234,10 @@ def is_request_body_safe( return True raise ValueError( f"Rejected Request: {param} is not allowed in request body. " - "Enable with `general_settings::allow_client_side_credentials` on proxy config.yaml. " + "Clientside passthrough requires explicit admin opt-in via " + "either `general_settings.allow_client_side_credentials = true` " + "(proxy-wide) or `configurable_clientside_auth_params` on the " + "deployment in your proxy config.yaml. " "Relevant Issue: https://huntr.com/bounties/4001e1a2-7b7a-4776-a3ae-e6692ec3d997", ) diff --git a/litellm/proxy/db/prisma_client.py b/litellm/proxy/db/prisma_client.py index 73735796eb3..d112e222307 100644 --- a/litellm/proxy/db/prisma_client.py +++ b/litellm/proxy/db/prisma_client.py @@ -52,18 +52,25 @@ class PrismaWrapper: engine = self._original_prisma._engine process = getattr(engine, "process", None) if engine is not None else None if process is not None: - return process.pid + pid = process.pid + if isinstance(pid, int): + return pid except (AttributeError, TypeError): pass return 0 @staticmethod async def _kill_engine_process(pid: int) -> None: - """Force-kill an orphaned engine subprocess to prevent DB connection pool leaks. + """Force-kill the engine subprocess to prevent DB connection pool leaks. - Called when disconnect() fails and the old engine process may still be - holding open connections. Sends SIGTERM for graceful shutdown, waits - briefly, then SIGKILL as a backstop. + Called on every reconnect (in `recreate_prisma_client`) to retire the + old query-engine subprocess without invoking prisma-client-py's + synchronous `disconnect()` — which blocks the asyncio event loop on + `subprocess.Popen.wait()` for 30-120+ seconds when the engine is + stuck on TCP close. + + Sends SIGTERM for graceful shutdown, waits briefly, then SIGKILL as + a backstop. """ if pid <= 0: return @@ -72,7 +79,7 @@ class PrismaWrapper: except (ProcessLookupError, PermissionError, OSError): return # Already dead or inaccessible verbose_proxy_logger.warning( - "Sent SIGTERM to orphaned prisma-query-engine PID %s after failed disconnect.", + "Sent SIGTERM to prisma-query-engine PID %s during reconnect.", pid, ) # Brief wait for graceful shutdown, then force-kill @@ -217,15 +224,18 @@ class PrismaWrapper: async def recreate_prisma_client( self, new_db_url: str, http_client: Optional[Any] = None ): - """Disconnect and reconnect the Prisma client with a new database URL.""" + """Disconnect and reconnect the Prisma client with a new database URL. + + Kills the old engine subprocess directly (SIGTERM → SIGKILL) rather than + calling `disconnect()`. prisma-client-py's `disconnect()` calls a + synchronous `subprocess.Popen.wait()` that can freeze the asyncio event + loop for 30-120+ seconds when the engine is stuck on TCP close, + breaking `/health/liveliness` and causing Kubernetes pod restarts. + """ from prisma import Prisma # type: ignore old_engine_pid = self._get_engine_pid() - - try: - await self._original_prisma.disconnect() - except Exception as e: - verbose_proxy_logger.warning(f"Failed to disconnect Prisma client: {e}") + if old_engine_pid > 0: await self._kill_engine_process(old_engine_pid) if http_client is not None: diff --git a/litellm/proxy/management_endpoints/workflow_management_endpoints.py b/litellm/proxy/management_endpoints/workflow_management_endpoints.py new file mode 100644 index 00000000000..a19af4dd484 --- /dev/null +++ b/litellm/proxy/management_endpoints/workflow_management_endpoints.py @@ -0,0 +1,492 @@ +""" +WORKFLOW RUN MANAGEMENT + +Generic durable state tracking for agents and automated workflows. + +POST /v1/workflows/runs - Create a workflow run +GET /v1/workflows/runs - List runs (filter by type, status) +GET /v1/workflows/runs/{run_id} - Get run with latest event +PATCH /v1/workflows/runs/{run_id} - Update status, metadata, output +POST /v1/workflows/runs/{run_id}/events - Append event (updates run status) +GET /v1/workflows/runs/{run_id}/events - Full event log +POST /v1/workflows/runs/{run_id}/messages - Append conversation message +GET /v1/workflows/runs/{run_id}/messages - Fetch conversation history +""" + +import json +from typing import Any, Dict, Literal, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query + +try: + from prisma.errors import UniqueViolationError +except ImportError: + UniqueViolationError = None # type: ignore +from pydantic import BaseModel + +from litellm._logging import verbose_proxy_logger +from litellm.proxy._types import CommonProxyErrors, LitellmUserRoles, UserAPIKeyAuth +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + +router = APIRouter() + +_MAX_SEQUENCE_RETRIES = 5 + + +def _json(value: Any) -> str: + """Serialize a Python value for prisma-client-py Json fields (must be a string).""" + return json.dumps(value) + + +def _is_admin(user_api_key_dict: UserAPIKeyAuth) -> bool: + return user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value + + +def _caller_key(user_api_key_dict: UserAPIKeyAuth) -> Optional[str]: + """Return the hashed key token that identifies this caller, or None for master key.""" + return user_api_key_dict.token + + +# Status transitions driven by event_type +_EVENT_STATUS_MAP: Dict[str, str] = { + "step.started": "running", + "step.failed": "failed", + "hook.waiting": "paused", + "hook.received": "running", +} + + +# --------------------------------------------------------------------------- +# Request / Response models +# --------------------------------------------------------------------------- + + +class WorkflowRunCreateRequest(BaseModel): + workflow_type: str + input: Optional[Dict[str, Any]] = None + metadata: Optional[Dict[str, Any]] = None + + +WorkflowRunStatus = Literal["pending", "running", "paused", "completed", "failed"] + + +class WorkflowRunUpdateRequest(BaseModel): + status: Optional[WorkflowRunStatus] = None + output: Optional[Dict[str, Any]] = None + metadata: Optional[Dict[str, Any]] = None + + +class WorkflowEventCreateRequest(BaseModel): + event_type: str + step_name: str + data: Optional[Dict[str, Any]] = None + + +class WorkflowMessageCreateRequest(BaseModel): + role: str + content: str + session_id: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def _get_next_sequence_number(prisma_client: Any, run_id: str, table: str) -> int: + """Return MAX(sequence_number) + 1 for the given run, for either events or messages.""" + if table == "events": + rows = await prisma_client.db.litellm_workflowevent.find_many( + where={"run_id": run_id}, + order={"sequence_number": "desc"}, + take=1, + ) + else: + rows = await prisma_client.db.litellm_workflowmessage.find_many( + where={"run_id": run_id}, + order={"sequence_number": "desc"}, + take=1, + ) + return (rows[0].sequence_number + 1) if rows else 0 + + +async def _require_run( + prisma_client: Any, + run_id: str, + user_api_key_dict: Optional[UserAPIKeyAuth] = None, +) -> Any: + """Return the run or raise 404. For non-admin callers, also enforce key ownership.""" + run = await prisma_client.db.litellm_workflowrun.find_unique( + where={"run_id": run_id} + ) + if run is None: + raise HTTPException(status_code=404, detail=f"Run '{run_id}' not found") + if user_api_key_dict is not None and not _is_admin(user_api_key_dict): + caller = _caller_key(user_api_key_dict) + if not caller or run.created_by != caller: + raise HTTPException(status_code=404, detail=f"Run '{run_id}' not found") + return run + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.post( + "/v1/workflows/runs", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def create_workflow_run( + data: WorkflowRunCreateRequest, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Create a new workflow run. Returns run_id and session_id. + + The caller's API key token is stored as created_by so that non-admin keys + can only see and modify their own runs. + """ + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + try: + create_data: Dict[str, Any] = { + "workflow_type": data.workflow_type, + "created_by": _caller_key(user_api_key_dict), + } + if data.input is not None: + create_data["input"] = _json(data.input) + if data.metadata is not None: + create_data["metadata"] = _json(data.metadata) + run = await prisma_client.db.litellm_workflowrun.create(data=create_data) + return run + except Exception as e: + verbose_proxy_logger.exception("Error creating workflow run: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.get( + "/v1/workflows/runs", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def list_workflow_runs( + workflow_type: Optional[str] = Query(None), + status: Optional[str] = Query(None), + limit: int = Query(50, ge=1, le=250), + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """List workflow runs. Filter by workflow_type and/or status. + + Non-admin callers only see runs created by their own API key. + """ + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + where: Dict[str, Any] = {} + if workflow_type: + where["workflow_type"] = workflow_type + if status: + statuses = [s.strip() for s in status.split(",")] + where["status"] = {"in": statuses} if len(statuses) > 1 else statuses[0] + + # Non-admin callers are scoped to their own key. + if not _is_admin(user_api_key_dict): + caller = _caller_key(user_api_key_dict) + if caller: + where["created_by"] = caller + + try: + runs = await prisma_client.db.litellm_workflowrun.find_many( + where=where, + order={"created_at": "desc"}, + take=limit, + ) + return {"runs": runs, "count": len(runs)} + except Exception as e: + verbose_proxy_logger.exception("Error listing workflow runs: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.get( + "/v1/workflows/runs/{run_id}", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def get_workflow_run( + run_id: str, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Get a workflow run with its most recent event.""" + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + try: + run = await prisma_client.db.litellm_workflowrun.find_unique( + where={"run_id": run_id}, + include={"events": {"order_by": {"sequence_number": "desc"}, "take": 1}}, + ) + if run is None: + raise HTTPException(status_code=404, detail=f"Run '{run_id}' not found") + if not _is_admin(user_api_key_dict): + caller = _caller_key(user_api_key_dict) + if not caller or run.created_by != caller: + raise HTTPException(status_code=404, detail=f"Run '{run_id}' not found") + return run + except HTTPException: + raise + except Exception as e: + verbose_proxy_logger.exception("Error getting workflow run: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.patch( + "/v1/workflows/runs/{run_id}", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def update_workflow_run( + run_id: str, + data: WorkflowRunUpdateRequest, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Update status, metadata, or output on a workflow run.""" + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + update: Dict[str, Any] = {} + if data.status is not None: + update["status"] = data.status + if data.output is not None: + update["output"] = _json(data.output) + if data.metadata is not None: + update["metadata"] = _json(data.metadata) + + if not update: + raise HTTPException(status_code=400, detail="No fields to update") + + # Enforce ownership before writing. + await _require_run(prisma_client, run_id, user_api_key_dict) + + try: + run = await prisma_client.db.litellm_workflowrun.update( + where={"run_id": run_id}, + data=update, + ) + if run is None: + raise HTTPException(status_code=404, detail=f"Run '{run_id}' not found") + return run + except HTTPException: + raise + except Exception as e: + verbose_proxy_logger.exception("Error updating workflow run: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post( + "/v1/workflows/runs/{run_id}/events", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def append_workflow_event( + run_id: str, + data: WorkflowEventCreateRequest, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Append an event to the run's event log. Also updates run.status if event_type maps to a status. + + Sequence numbers use optimistic concurrency: on a unique-constraint collision + (concurrent append), retries up to _MAX_SEQUENCE_RETRIES times with a fresh MAX+1. + The event+status update is atomic in a single DB transaction. + """ + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + await _require_run(prisma_client, run_id, user_api_key_dict) + + new_status = _EVENT_STATUS_MAP.get(data.event_type) + + for attempt in range(_MAX_SEQUENCE_RETRIES): + try: + seq = await _get_next_sequence_number(prisma_client, run_id, "events") + event_data: Dict[str, Any] = { + "run_id": run_id, + "event_type": data.event_type, + "step_name": data.step_name, + "sequence_number": seq, + } + if data.data is not None: + event_data["data"] = _json(data.data) + + async with prisma_client.db.tx() as tx: + event = await tx.litellm_workflowevent.create(data=event_data) + if new_status: + await tx.litellm_workflowrun.update( + where={"run_id": run_id}, + data={"status": new_status}, + ) + + return event + + except Exception as e: + if UniqueViolationError is not None and isinstance(e, UniqueViolationError): + if attempt == _MAX_SEQUENCE_RETRIES - 1: + verbose_proxy_logger.exception( + "Sequence number collision after %d retries for run %s", + _MAX_SEQUENCE_RETRIES, + run_id, + ) + raise HTTPException( + status_code=409, + detail="Concurrent write conflict — please retry", + ) + continue + verbose_proxy_logger.exception("Error appending workflow event: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + raise HTTPException( + status_code=500, detail="Failed to append event" + ) # pragma: no cover + + +@router.get( + "/v1/workflows/runs/{run_id}/events", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def list_workflow_events( + run_id: str, + limit: int = Query(100, ge=1, le=500), + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Fetch event log for a run, ordered by sequence_number. Default limit 100, max 500.""" + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + await _require_run(prisma_client, run_id, user_api_key_dict) + + try: + events = await prisma_client.db.litellm_workflowevent.find_many( + where={"run_id": run_id}, + order={"sequence_number": "asc"}, + take=limit, + ) + return {"events": events, "count": len(events)} + except Exception as e: + verbose_proxy_logger.exception("Error listing workflow events: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post( + "/v1/workflows/runs/{run_id}/messages", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def append_workflow_message( + run_id: str, + data: WorkflowMessageCreateRequest, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Append a conversation message. Stores full content (not truncated). + + Uses optimistic concurrency for sequence numbers. + """ + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + await _require_run(prisma_client, run_id, user_api_key_dict) + + for attempt in range(_MAX_SEQUENCE_RETRIES): + try: + seq = await _get_next_sequence_number(prisma_client, run_id, "messages") + msg_data: Dict[str, Any] = { + "run_id": run_id, + "role": data.role, + "content": data.content, + "sequence_number": seq, + } + if data.session_id is not None: + msg_data["session_id"] = data.session_id + msg = await prisma_client.db.litellm_workflowmessage.create(data=msg_data) + return msg + + except Exception as e: + if UniqueViolationError is not None and isinstance(e, UniqueViolationError): + if attempt == _MAX_SEQUENCE_RETRIES - 1: + verbose_proxy_logger.exception( + "Sequence number collision after %d retries for run %s", + _MAX_SEQUENCE_RETRIES, + run_id, + ) + raise HTTPException( + status_code=409, + detail="Concurrent write conflict — please retry", + ) + continue + verbose_proxy_logger.exception("Error appending workflow message: %s", e) + raise HTTPException(status_code=500, detail=str(e)) + + raise HTTPException( + status_code=500, detail="Failed to append message" + ) # pragma: no cover + + +@router.get( + "/v1/workflows/runs/{run_id}/messages", + tags=["workflow management"], + dependencies=[Depends(user_api_key_auth)], +) +async def list_workflow_messages( + run_id: str, + limit: int = Query(100, ge=1, le=500), + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +): + """Fetch conversation history for a run, ordered by sequence_number. Default limit 100, max 500.""" + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + await _require_run(prisma_client, run_id, user_api_key_dict) + + try: + messages = await prisma_client.db.litellm_workflowmessage.find_many( + where={"run_id": run_id}, + order={"sequence_number": "asc"}, + take=limit, + ) + return {"messages": messages, "count": len(messages)} + except Exception as e: + verbose_proxy_logger.exception("Error listing workflow messages: %s", e) + raise HTTPException(status_code=500, detail=str(e)) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 6450a16df53..6e7e98205d5 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -427,6 +427,9 @@ from litellm.proxy.management_endpoints.team_endpoints import ( from litellm.proxy.management_endpoints.tool_management_endpoints import ( router as tool_management_router, ) +from litellm.proxy.management_endpoints.workflow_management_endpoints import ( + router as workflow_management_router, +) from litellm.proxy.memory.memory_endpoints import router as memory_router from litellm.proxy.management_endpoints.ui_sso import ( get_disabled_non_admin_personal_key_creation, @@ -14285,6 +14288,7 @@ app.include_router(model_management_router) app.include_router(model_access_group_management_router) app.include_router(tag_management_router) app.include_router(tool_management_router) +app.include_router(workflow_management_router) app.include_router(memory_router) app.include_router(cost_tracking_settings_router) app.include_router(router_settings_router) diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 8f07c5afa3f..fd54ed5243a 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1290,3 +1290,80 @@ model LiteLLM_AdaptiveRouterSession { @@id([session_id, router_name, model_name]) @@index([last_activity_at], map: "idx_adaptive_router_session_activity") } + +// --------------------------------------------------------------------------- +// Workflow Run Tracking +// +// Generic durable state tracking for any agent or automated workflow. +// Design: three tables — run (header + materialized status), event (append-only +// source of truth for state transitions), message (conversation inbox/outbox). +// +// Usage: +// - Set `workflow_type` to identify the owning system (e.g. "shin-builder"). +// - Store domain-specific fields in `metadata` (worktree_path, pr_url, etc.). +// - `session_id` on WorkflowRun matches `x-litellm-session-id` header sent to +// the proxy — all spend logs for this run are automatically tagged. +// --------------------------------------------------------------------------- + +// One instance of work being done. `status` is a materialized cache of the +// latest event; the event log is the authoritative source of truth. +model LiteLLM_WorkflowRun { + run_id String @id @default(uuid()) + session_id String @unique @default(uuid()) + workflow_type String + status String @default("pending") + created_by String? // user_id of the key that created this run; null = created by master key + created_at DateTime @default(now()) + updated_at DateTime @updatedAt + input Json? + output Json? + metadata Json? + + events LiteLLM_WorkflowEvent[] + messages LiteLLM_WorkflowMessage[] + + @@index([workflow_type, status]) + @@index([session_id]) + @@index([created_at]) + @@index([created_by]) +} + +// Append-only log of state transitions. Never mutate rows here. +// `step_name` and `event_type` are caller-defined strings — no hardcoded enums. +// Status auto-update rules (applied by the append endpoint): +// step.started → run.status = running +// step.failed → run.status = failed +// hook.waiting → run.status = paused +// hook.received → run.status = running +model LiteLLM_WorkflowEvent { + event_id String @id @default(uuid()) + run_id String + event_type String + step_name String + sequence_number Int + data Json? + created_at DateTime @default(now()) + + run LiteLLM_WorkflowRun @relation(fields: [run_id], references: [run_id]) + + @@unique([run_id, sequence_number]) + @@index([run_id]) +} + +// Conversation inbox/outbox — full message content, separate from the durable +// event log. Spend logs truncate messages; this table stores them in full. +// `session_id` here is the Claude --resume session ID (or similar). +model LiteLLM_WorkflowMessage { + message_id String @id @default(uuid()) + run_id String + role String + content String + sequence_number Int + session_id String? + created_at DateTime @default(now()) + + run LiteLLM_WorkflowRun @relation(fields: [run_id], references: [run_id]) + + @@unique([run_id, sequence_number]) + @@index([run_id]) +} diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 93cc8bfd22b..d2dfa177515 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -4198,8 +4198,11 @@ class PrismaClient: Uses the _engine_confirmed_dead flag (set by waitpid thread / pidfd / poll handlers) to choose between heavy reconnect (engine dead -- recreate - Prisma client, re-arm watcher) and lightweight reconnect (network - blip -- disconnect, connect, SELECT 1). + Prisma client, re-arm watcher) and direct reconnect (network blip -- + recreate Prisma client, re-arm watcher, SELECT 1). Both paths recreate + the client via the non-blocking kill-then-construct flow rather than + calling disconnect(), which blocks the event loop on the synchronous + subprocess.Popen.wait() inside prisma-client-py (see issue #26191). """ effective_timeout = ( timeout_seconds @@ -4243,17 +4246,20 @@ class PrismaClient: ) async def _do_direct_reconnect() -> None: - old_pid = self._get_engine_pid() - try: - await self.db.disconnect() - except Exception as disconnect_err: - verbose_proxy_logger.warning( - "Prisma DB disconnect before reconnect failed: %s", - disconnect_err, + db_url = os.getenv("DATABASE_URL", "") + if not db_url: + verbose_proxy_logger.error( + "DATABASE_URL not set; cannot reconnect Prisma client." ) - await PrismaWrapper._kill_engine_process(old_pid) - - await self.db.connect() + raise RuntimeError("DATABASE_URL not set") + # Fresh Prisma client + new engine subprocess. The previous + # "lightweight" path called `disconnect()` which blocks the + # event loop on `subprocess.Popen.wait()`; since that call + # ends up killing the engine anyway, we do it non-blockingly + # via `_kill_engine_process` inside `recreate_prisma_client`. + self._cleanup_engine_watcher() + await self.db.recreate_prisma_client(db_url) + await self._start_engine_watcher() await self.db.query_raw("SELECT 1") await asyncio.wait_for(_do_direct_reconnect(), timeout=effective_timeout) diff --git a/litellm/proxy/workflows/README.md b/litellm/proxy/workflows/README.md new file mode 100644 index 00000000000..f452066afb0 --- /dev/null +++ b/litellm/proxy/workflows/README.md @@ -0,0 +1,150 @@ +# Workflow Run Tracking + +Generic durable state tracking for agents and automated workflows built on the LiteLLM proxy. + +## The Problem + +Agents like [shin-builder](https://github.com/BerriAI/shin-builder) run multi-stage pipelines (triage → plan → implement → PR). Their task state and conversation history lived in memory — a process restart lost everything. + +## Three-Table Design + +``` +WorkflowRun one instance of work (header + materialized status) +WorkflowEvent append-only state transitions (source of truth for replay) +WorkflowMessage conversation inbox/outbox (full content, not truncated) +``` + +**WorkflowEvent is the source of truth.** `WorkflowRun.status` is a materialized cache updated automatically when events are appended. If you need to debug a run, replay its events. + +## API + +All endpoints require a valid LiteLLM API key (`Authorization: Bearer sk-...`). + +### Runs + +``` +POST /v1/workflows/runs Create a run +GET /v1/workflows/runs List runs (?workflow_type=&status=) +GET /v1/workflows/runs/{run_id} Get run + latest event +PATCH /v1/workflows/runs/{run_id} Update status / metadata / output +``` + +### Events + +``` +POST /v1/workflows/runs/{run_id}/events Append event (auto-updates run status) +GET /v1/workflows/runs/{run_id}/events Full event log (ordered by sequence) +``` + +### Messages + +``` +POST /v1/workflows/runs/{run_id}/messages Append message +GET /v1/workflows/runs/{run_id}/messages Conversation history (ordered by sequence) +``` + +## Quick Start + +```bash +# Create a run +curl -X POST http://localhost:4000/v1/workflows/runs \ + -H "Authorization: Bearer sk-1234" \ + -H "Content-Type: application/json" \ + -d '{"workflow_type": "shin-builder", "metadata": {"title": "Fix login bug"}}' + +# {"run_id": "abc-123", "session_id": "xyz-456", "status": "pending", ...} + +# Mark step started (sets status → running) +curl -X POST http://localhost:4000/v1/workflows/runs/abc-123/events \ + -H "Authorization: Bearer sk-1234" \ + -H "Content-Type: application/json" \ + -d '{"event_type": "step.started", "step_name": "grill", "data": {"claude_session_id": "sess-789"}}' + +# Store a conversation message +curl -X POST http://localhost:4000/v1/workflows/runs/abc-123/messages \ + -H "Authorization: Bearer sk-1234" \ + -H "Content-Type: application/json" \ + -d '{"role": "user", "content": "What is the expected behavior?", "session_id": "sess-789"}' + +# Restart recovery: fetch active runs and resume from last event's data.claude_session_id +curl "http://localhost:4000/v1/workflows/runs?status=running,paused&workflow_type=shin-builder" \ + -H "Authorization: Bearer sk-1234" +``` + +## Status Auto-Update Rules + +When you append an event, the run's status is updated automatically: + +| event_type | run.status | +|-----------------|------------| +| `step.started` | `running` | +| `step.failed` | `failed` | +| `hook.waiting` | `paused` | +| `hook.received` | `running` | + +Set `status = completed` explicitly via PATCH when the workflow finishes. + +## Linking to Spend Logs + +`WorkflowRun.session_id` is generated automatically (UUID). Pass it as the `x-litellm-session-id` header when making completions through the proxy: + +```python +headers = {"x-litellm-session-id": run.session_id} +``` + +All spend log entries for this run are then tagged automatically. Query cost per run: + +``` +POST /ui/spend_logs/view_session_spend_logs?session_id={run.session_id} +``` + +## Sequence Numbers + +Sequence numbers on events and messages are assigned server-side (`MAX + 1` per run). Callers never supply them. This guarantees ordering even under concurrent writes. + +## Using from shin-builder + +Replace the in-memory `tasks.py` dict with calls to these endpoints: + +```python +import httpx + +class WorkflowRunClient: + def __init__(self, base_url: str, api_key: str): + self._client = httpx.AsyncClient( + base_url=base_url, + headers={"Authorization": f"Bearer {api_key}"}, + ) + + async def create_task(self, title: str, **metadata) -> dict: + r = await self._client.post("/v1/workflows/runs", json={ + "workflow_type": "shin-builder", + "metadata": {"title": title, **metadata}, + }) + r.raise_for_status() + return r.json() + + async def list_active_tasks(self) -> list: + r = await self._client.get( + "/v1/workflows/runs", + params={"workflow_type": "shin-builder", "status": "running,paused"}, + ) + r.raise_for_status() + return r.json()["runs"] + + async def transition(self, run_id: str, step_name: str, event_type: str, data: dict = None): + r = await self._client.post(f"/v1/workflows/runs/{run_id}/events", json={ + "event_type": event_type, + "step_name": step_name, + "data": data or {}, + }) + r.raise_for_status() + + async def append_message(self, run_id: str, role: str, content: str, session_id: str = None): + r = await self._client.post(f"/v1/workflows/runs/{run_id}/messages", json={ + "role": role, "content": content, "session_id": session_id, + }) + r.raise_for_status() +``` + +On startup, call `list_active_tasks()` to restore in-flight runs. The last `step.started` event's `data.claude_session_id` gives you the `--resume` ID. diff --git a/litellm/router_utils/clientside_credential_handler.py b/litellm/router_utils/clientside_credential_handler.py index c98f614335b..45ade81b2dd 100644 --- a/litellm/router_utils/clientside_credential_handler.py +++ b/litellm/router_utils/clientside_credential_handler.py @@ -11,9 +11,60 @@ If given, generate a unique model_id for the deployment. Ensures cooldowns are applied correctly. """ +from typing import List + clientside_credential_keys = ["api_key", "api_base", "base_url"] +def _admin_config_fields_to_clear_on_base_override() -> List[str]: + """ + Provider-specific credential / endpoint-targeting fields that must NOT + flow through to a client-redirected upstream. + + Built dynamically from ``CredentialLiteLLMParams.model_fields`` so any + new provider field added there (Bedrock endpoint, Watsonx region, etc.) + is gated automatically — plus a fixed list of kwargs-only fields that + aren't declared on the typed model. + """ + from litellm.types.router import CredentialLiteLLMParams + + typed_fields = [ + f + for f in CredentialLiteLLMParams.model_fields + if f not in clientside_credential_keys + ] + kwargs_only_fields = [ + # Caller-supplied via **kwargs, not declared on CredentialLiteLLMParams. + "organization", + "extra_body", + "extra_headers", + "default_headers", + "api_type", + "azure_ad_token", + "azure_ad_token_provider", + "aws_session_token", + "aws_sts_endpoint", + "aws_web_identity_token", + "aws_role_name", + # OCI provider — consumed by litellm/llms/oci/* via optional_params + # and not declared on CredentialLiteLLMParams. Without these here, + # an admin's OCI signing key / tenancy / fingerprint would flow + # through to an attacker-redirected upstream. + "oci_signer", + "oci_user", + "oci_fingerprint", + "oci_tenancy", + "oci_key", + "oci_key_file", + ] + return typed_fields + kwargs_only_fields + + +_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE = ( + _admin_config_fields_to_clear_on_base_override() +) + + def is_clientside_credential(request_kwargs: dict) -> bool: """ Check if the credential is a clientside credential. @@ -34,4 +85,20 @@ def get_dynamic_litellm_params(litellm_params: dict, request_kwargs: dict) -> di for key in clientside_credential_keys: if key in request_kwargs: litellm_params[key] = request_kwargs[key] + + # If the caller redirected api_base/base_url to a client-controlled value, + # don't forward the admin's organization / extra_body / region / token / + # vertex / aws fields — those were meant for the original upstream. + # Always drop the admin's value first, then write the caller's value back + # if they resupplied the field. The naive + # ``if field not in request_kwargs: pop`` shape lets a caller *echo* a + # field name (with any value, including an empty string) to keep the + # admin's value in ``litellm_params`` and have it forwarded to the + # redirected upstream. + if "api_base" in request_kwargs or "base_url" in request_kwargs: + for field in _ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE: + litellm_params.pop(field, None) + if field in request_kwargs: + litellm_params[field] = request_kwargs[field] + return litellm_params diff --git a/schema.prisma b/schema.prisma index 8f07c5afa3f..fd54ed5243a 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1290,3 +1290,80 @@ model LiteLLM_AdaptiveRouterSession { @@id([session_id, router_name, model_name]) @@index([last_activity_at], map: "idx_adaptive_router_session_activity") } + +// --------------------------------------------------------------------------- +// Workflow Run Tracking +// +// Generic durable state tracking for any agent or automated workflow. +// Design: three tables — run (header + materialized status), event (append-only +// source of truth for state transitions), message (conversation inbox/outbox). +// +// Usage: +// - Set `workflow_type` to identify the owning system (e.g. "shin-builder"). +// - Store domain-specific fields in `metadata` (worktree_path, pr_url, etc.). +// - `session_id` on WorkflowRun matches `x-litellm-session-id` header sent to +// the proxy — all spend logs for this run are automatically tagged. +// --------------------------------------------------------------------------- + +// One instance of work being done. `status` is a materialized cache of the +// latest event; the event log is the authoritative source of truth. +model LiteLLM_WorkflowRun { + run_id String @id @default(uuid()) + session_id String @unique @default(uuid()) + workflow_type String + status String @default("pending") + created_by String? // user_id of the key that created this run; null = created by master key + created_at DateTime @default(now()) + updated_at DateTime @updatedAt + input Json? + output Json? + metadata Json? + + events LiteLLM_WorkflowEvent[] + messages LiteLLM_WorkflowMessage[] + + @@index([workflow_type, status]) + @@index([session_id]) + @@index([created_at]) + @@index([created_by]) +} + +// Append-only log of state transitions. Never mutate rows here. +// `step_name` and `event_type` are caller-defined strings — no hardcoded enums. +// Status auto-update rules (applied by the append endpoint): +// step.started → run.status = running +// step.failed → run.status = failed +// hook.waiting → run.status = paused +// hook.received → run.status = running +model LiteLLM_WorkflowEvent { + event_id String @id @default(uuid()) + run_id String + event_type String + step_name String + sequence_number Int + data Json? + created_at DateTime @default(now()) + + run LiteLLM_WorkflowRun @relation(fields: [run_id], references: [run_id]) + + @@unique([run_id, sequence_number]) + @@index([run_id]) +} + +// Conversation inbox/outbox — full message content, separate from the durable +// event log. Spend logs truncate messages; this table stores them in full. +// `session_id` here is the Claude --resume session ID (or similar). +model LiteLLM_WorkflowMessage { + message_id String @id @default(uuid()) + run_id String + role String + content String + sequence_number Int + session_id String? + created_at DateTime @default(now()) + + run LiteLLM_WorkflowRun @relation(fields: [run_id], references: [run_id]) + + @@unique([run_id, sequence_number]) + @@index([run_id]) +} diff --git a/tests/batches_tests/test_openai_batches_and_files.py b/tests/batches_tests/test_openai_batches_and_files.py index 0aed224c256..a64f208b3f1 100644 --- a/tests/batches_tests/test_openai_batches_and_files.py +++ b/tests/batches_tests/test_openai_batches_and_files.py @@ -557,6 +557,7 @@ async def test_avertex_batch_prediction(monkeypatch): mock_get_response = MagicMock() mock_get_response.json.return_value = mock_vertex_batch_response mock_get_response.status_code = 200 + mock_get_response.is_redirect = False mock_get_response.raise_for_status.return_value = None mock_get.return_value = mock_get_response diff --git a/tests/litellm/proxy/test_prisma_engine_watchdog.py b/tests/litellm/proxy/test_prisma_engine_watchdog.py index 786167b9486..0d241f75749 100644 --- a/tests/litellm/proxy/test_prisma_engine_watchdog.py +++ b/tests/litellm/proxy/test_prisma_engine_watchdog.py @@ -254,32 +254,48 @@ async def test_run_reconnect_cycle_uses_heavy_path_when_confirmed_dead( @pytest.mark.asyncio -async def test_run_reconnect_cycle_uses_lightweight_path_when_engine_alive( +async def test_run_reconnect_cycle_uses_direct_path_when_engine_alive( engine_client, ) -> None: - """_run_reconnect_cycle uses disconnect/connect when engine is alive.""" - engine_client._engine_pid = 1234 + """Direct reconnect (engine alive) calls recreate_prisma_client + SELECT 1. - with patch.object(engine_client, "_is_engine_alive", return_value=True): + The old "lightweight" path called `disconnect()` + `connect()`, which + blocks the event loop on the sync `process.wait()` inside aclose(). + The fix routes both engine-alive and engine-dead paths through + `recreate_prisma_client`, which non-blockingly kills the old engine. + """ + engine_client._engine_pid = 1234 + engine_client._start_engine_watcher = AsyncMock() + + with ( + patch.object(engine_client, "_is_engine_alive", return_value=True), + patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}), + ): await engine_client._run_reconnect_cycle(timeout_seconds=5.0) - engine_client.db.connect.assert_awaited_once() + engine_client.db.recreate_prisma_client.assert_awaited_once_with( + "postgresql://test" + ) engine_client.db.query_raw.assert_awaited_once_with("SELECT 1") - engine_client.db.recreate_prisma_client.assert_not_awaited() + engine_client.db.disconnect.assert_not_awaited() @pytest.mark.asyncio -async def test_run_reconnect_cycle_uses_lightweight_path_when_pid_unknown( +async def test_run_reconnect_cycle_uses_direct_path_when_pid_unknown( engine_client, ) -> None: - """_run_reconnect_cycle uses lightweight path when engine PID is not tracked.""" + """When the engine PID is not tracked, direct reconnect still runs.""" engine_client._engine_pid = 0 + engine_client._start_engine_watcher = AsyncMock() - await engine_client._run_reconnect_cycle(timeout_seconds=5.0) + with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}): + await engine_client._run_reconnect_cycle(timeout_seconds=5.0) - engine_client.db.connect.assert_awaited_once() + engine_client.db.recreate_prisma_client.assert_awaited_once_with( + "postgresql://test" + ) engine_client.db.query_raw.assert_awaited_once_with("SELECT 1") - engine_client.db.recreate_prisma_client.assert_not_awaited() + engine_client.db.disconnect.assert_not_awaited() @pytest.mark.asyncio @@ -473,36 +489,38 @@ def test_on_engine_death_from_thread_ignores_stale_pid(engine_client): @pytest.mark.asyncio -async def test_escalation_after_consecutive_lightweight_failures(engine_client): - """After N consecutive lightweight reconnect failures, _engine_confirmed_dead +async def test_escalation_after_consecutive_direct_reconnect_failures(engine_client): + """After N consecutive direct reconnect failures, _engine_confirmed_dead is set to True so _run_reconnect_cycle takes the heavy reconnect path.""" engine_client._reconnect_escalation_threshold = 3 engine_client._consecutive_reconnect_failures = 0 engine_client._db_reconnect_cooldown_seconds = 0 # disable cooldown for test + engine_client._start_engine_watcher = AsyncMock(return_value=None) - # Make lightweight reconnect fail every time - engine_client.db.disconnect = AsyncMock(return_value=None) - engine_client.db.connect = AsyncMock(side_effect=Exception("connect failed")) + # Make direct reconnect fail every time + engine_client.db.recreate_prisma_client = AsyncMock( + side_effect=Exception("recreate failed") + ) # Run 3 failed reconnect attempts - for i in range(3): - result = await engine_client._attempt_reconnect_inside_lock( - force=True, reason="test", timeout_seconds=5.0 - ) - assert result is False + with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}): + for _ in range(3): + result = await engine_client._attempt_reconnect_inside_lock( + force=True, reason="test", timeout_seconds=5.0 + ) + assert result is False assert engine_client._consecutive_reconnect_failures == 3 - # Next attempt should escalate: _engine_confirmed_dead set to True before _run_reconnect_cycle + # Next attempt should escalate to the heavy path (recreate_prisma_client still + # the call, but via the _engine_confirmed_dead branch that also re-arms the watcher). engine_client.db.recreate_prisma_client = AsyncMock(return_value=None) - engine_client._start_engine_watcher = AsyncMock(return_value=None) with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}): result = await engine_client._attempt_reconnect_inside_lock( force=True, reason="test_escalation", timeout_seconds=5.0 ) - # Heavy reconnect should have been attempted (recreate_prisma_client called) engine_client.db.recreate_prisma_client.assert_awaited_once() @@ -511,15 +529,16 @@ async def test_successful_reconnect_resets_failure_counter(engine_client): """A successful reconnect resets _consecutive_reconnect_failures to 0.""" engine_client._consecutive_reconnect_failures = 2 engine_client._db_reconnect_cooldown_seconds = 0 + engine_client._start_engine_watcher = AsyncMock() # Make reconnect succeed - engine_client.db.disconnect = AsyncMock(return_value=None) - engine_client.db.connect = AsyncMock(return_value=None) + engine_client.db.recreate_prisma_client = AsyncMock(return_value=None) engine_client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) - result = await engine_client._attempt_reconnect_inside_lock( - force=True, reason="test", timeout_seconds=5.0 - ) + with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}): + result = await engine_client._attempt_reconnect_inside_lock( + force=True, reason="test", timeout_seconds=5.0 + ) assert result is True assert engine_client._consecutive_reconnect_failures == 0 diff --git a/tests/test_litellm/litellm_core_utils/test_get_llm_provider_endpoint_match.py b/tests/test_litellm/litellm_core_utils/test_get_llm_provider_endpoint_match.py new file mode 100644 index 00000000000..fc5b39a2fd7 --- /dev/null +++ b/tests/test_litellm/litellm_core_utils/test_get_llm_provider_endpoint_match.py @@ -0,0 +1,138 @@ +""" +Regression tests for the parsed-URL hostname match used to identify a +caller-supplied ``api_base`` as a known openai-compatible provider. + +The previous shape (``if endpoint in api_base:``) used unanchored +substring search, which let a caller pass +``https://attacker.com/api.groq.com/openai/v1`` and have the proxy +return ``GROQ_API_KEY`` as the dynamic credential — exfiltrating the +server's real provider key to an attacker-controlled host on the +outbound request. +""" + +import os +import sys +from unittest.mock import patch + +import pytest + +sys.path.insert(0, os.path.abspath("../../..")) + +from litellm.litellm_core_utils.get_llm_provider_logic import ( + _endpoint_matches_api_base, + get_llm_provider, +) + + +class TestEndpointMatchesApiBase: + """Direct unit tests on the parsed-URL matcher.""" + + @pytest.mark.parametrize( + "endpoint, api_base", + [ + # Bare hostname endpoint, exact host match. + ("api.perplexity.ai", "https://api.perplexity.ai/v1"), + # Endpoint includes a path; api_base path starts with it. + ("api.groq.com/openai/v1", "https://api.groq.com/openai/v1"), + # Endpoint with full URL scheme. + ("https://api.cerebras.ai/v1", "https://api.cerebras.ai/v1/chat"), + # Trailing-slash on registered endpoint must not break match. + ("https://llm.chutes.ai/v1/", "https://llm.chutes.ai/v1/chat"), + # Case-insensitive on hostname. + ("api.groq.com/openai/v1", "https://API.GROQ.COM/openai/v1"), + ], + ) + def test_legitimate_provider_urls_match(self, endpoint, api_base): + assert _endpoint_matches_api_base(endpoint, api_base) is True + + @pytest.mark.parametrize( + "endpoint, api_base", + [ + # Attacker host, registered endpoint smuggled into path. + ( + "api.groq.com/openai/v1", + "https://attacker.com/api.groq.com/openai/v1", + ), + # Attacker host, registered endpoint smuggled into a path segment. + ( + "api.groq.com/openai/v1", + "https://attacker.com/foo/api.groq.com/openai/v1", + ), + # Lookalike host that contains the registered host as a suffix label. + ( + "api.groq.com/openai/v1", + "https://api.groq.com.attacker.com/openai/v1", + ), + # Lookalike host with the registered host as a prefix. + ( + "api.groq.com/openai/v1", + "https://api.groq.com.evil.example/openai/v1", + ), + # Right host, wrong path — endpoint requires ``/openai/v1`` prefix. + ("api.groq.com/openai/v1", "https://api.groq.com/v1"), + # Path-segment lookalike: ``/openai/v10`` must not match ``/openai/v1``. + ("api.groq.com/openai/v1", "https://api.groq.com/openai/v10"), + # Userinfo / @-injection trick — the ``hostname`` after ``@`` is + # what httpx connects to. + ( + "api.groq.com/openai/v1", + "https://api.groq.com@attacker.com/openai/v1", + ), + ], + ) + def test_attacker_smuggling_does_not_match(self, endpoint, api_base): + assert _endpoint_matches_api_base(endpoint, api_base) is False + + +class TestGetLlmProviderRejectsAttackerSmuggledApiBase: + """ + End-to-end: ``get_llm_provider`` must NOT return the server's stored + secret (e.g. ``GROQ_API_KEY``) for an api_base whose hostname is + attacker-controlled, even when the registered endpoint string appears + elsewhere in the URL. + """ + + def test_attacker_host_does_not_yield_groq_secret(self): + # The function may either fall through (different provider) or + # raise BadRequestError because the model can't be identified. + # The invariant under test is that ``GROQ_API_KEY`` is never + # looked up against an attacker-controlled hostname. + import litellm + + with patch( + "litellm.litellm_core_utils.get_llm_provider_logic.get_secret_str", + return_value="server-real-groq-key", + ) as mocked_secret: + try: + _, _, dynamic_api_key, _ = get_llm_provider( + model="some-model", + api_base="https://attacker.com/api.groq.com/openai/v1", + ) + # If it returned, the dynamic key must not be the secret. + assert dynamic_api_key != "server-real-groq-key" + except litellm.exceptions.BadRequestError: + # Acceptable outcome: provider unidentifiable, no secret + # was returned. + pass + + # Regardless of return / raise, the secret must never have been + # read against this attacker-controlled api_base. + groq_lookups = [ + call + for call in mocked_secret.call_args_list + if call.args and call.args[0] == "GROQ_API_KEY" + ] + assert groq_lookups == [] + + def test_legitimate_groq_api_base_still_resolves(self): + with patch( + "litellm.litellm_core_utils.get_llm_provider_logic.get_secret_str", + return_value="server-real-groq-key", + ): + _, provider, dynamic_api_key, _ = get_llm_provider( + model="some-model", + api_base="https://api.groq.com/openai/v1", + ) + + assert provider == "groq" + assert dynamic_api_key == "server-real-groq-key" diff --git a/tests/test_litellm/proxy/auth/test_auth_utils.py b/tests/test_litellm/proxy/auth/test_auth_utils.py index 15cfc84c232..91f300b88ce 100644 --- a/tests/test_litellm/proxy/auth/test_auth_utils.py +++ b/tests/test_litellm/proxy/auth/test_auth_utils.py @@ -5,6 +5,8 @@ Unit tests for auth_utils functions related to rate limiting and customer ID ext from typing import Optional from unittest.mock import MagicMock, patch +import pytest + from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.auth.auth_utils import ( _get_customer_id_from_standard_headers, @@ -15,6 +17,7 @@ from litellm.proxy.auth.auth_utils import ( get_key_model_tpm_limit, get_project_model_rpm_limit, get_project_model_tpm_limit, + is_request_body_safe, ) @@ -660,3 +663,304 @@ class TestCheckCompleteCredentials: def test_returns_true_when_api_key_is_valid(self): result = check_complete_credentials({"model": "gpt-4", "api_key": "sk-valid"}) assert result is True + + +class TestCheckCompleteCredentialsBlocksSSRF: + """ + Even with credentials supplied, ``api_base`` / ``base_url`` must not + point at private / internal / cloud-metadata addresses. Without this + the gate accepts ``api_key=anything`` plus a malicious target and the + proxy is used as an SSRF pivot. + + The check only runs when ``litellm.user_url_validation`` is True, so + every test in this class flips the toggle. Tests stay mock-only — no + real DNS is performed. + """ + + @pytest.fixture(autouse=True) + def _enable_url_validation(self, monkeypatch): + import litellm + + monkeypatch.setattr(litellm, "user_url_validation", True, raising=False) + + @pytest.mark.parametrize( + "url_field", + ["api_base", "base_url"], + ) + @pytest.mark.parametrize( + "blocked_url", + [ + "http://169.254.169.254/latest/meta-data/iam/security-credentials/", + "http://metadata.google.internal/computeMetadata/v1/", + "http://127.0.0.1:8080/admin", + "http://10.0.0.1/", + "http://192.168.1.1/", + ], + ) + def test_rejects_private_or_metadata_targets(self, url_field, blocked_url): + from litellm.litellm_core_utils.url_utils import SSRFError + + with patch( + "litellm.proxy.auth.auth_utils.validate_url", + side_effect=SSRFError(f"blocked: {blocked_url}"), + ): + with pytest.raises(ValueError) as exc_info: + check_complete_credentials( + { + "model": "gpt-4", + "api_key": "sk-some-clientside-key", + url_field: blocked_url, + } + ) + assert url_field in str(exc_info.value) + assert "SSRF" in str(exc_info.value) + + def test_allows_public_target_when_validate_url_passes(self): + # ``validate_url`` is mocked so no real DNS is performed. + with patch( + "litellm.proxy.auth.auth_utils.validate_url", + return_value=("https://api.openai.com/v1", "api.openai.com"), + ): + result = check_complete_credentials( + { + "model": "gpt-4", + "api_key": "sk-some-clientside-key", + "api_base": "https://api.openai.com/v1", + } + ) + assert result is True + + def test_skips_url_validation_when_toggle_is_off(self, monkeypatch): + # Admins who disable ``user_url_validation`` (default) should not + # have requests rejected at the proxy boundary even if the URL + # would fail the SSRF guard. + import litellm + + monkeypatch.setattr(litellm, "user_url_validation", False, raising=False) + with patch( + "litellm.proxy.auth.auth_utils.validate_url", + ) as mocked: + result = check_complete_credentials( + { + "model": "gpt-4", + "api_key": "sk-some-clientside-key", + "api_base": "http://127.0.0.1:8080/admin", + } + ) + assert result is True + mocked.assert_not_called() + + +class TestGetDynamicLitellmParamsClearsAdminConfigOnBaseOverride: + """ + When the caller redirects ``api_base`` / ``base_url`` to their own + server, admin-set fields like ``OpenAI-Organization``, ``extra_body``, + AWS / Vertex / Azure tokens, and per-deployment ``api_version`` must + NOT flow through to that destination. + """ + + def test_clears_admin_organization_and_extra_body_on_base_override(self): + from litellm.router_utils.clientside_credential_handler import ( + get_dynamic_litellm_params, + ) + + admin_params = { + "model": "gpt-4", + "api_key": "sk-admin-key", + "api_base": "https://admin.upstream/v1", + "organization": "org-admin-corp", + "extra_body": {"x-admin-secret": "super-secret"}, + "api_version": "2026-04-01", + } + out = get_dynamic_litellm_params( + litellm_params=dict(admin_params), + request_kwargs={ + "api_key": "sk-attacker", + "api_base": "https://attacker.example", + }, + ) + assert out["api_base"] == "https://attacker.example" + assert out["api_key"] == "sk-attacker" + assert "organization" not in out + assert "extra_body" not in out + assert "api_version" not in out + + def test_clears_aws_and_vertex_secrets_on_base_override(self): + from litellm.router_utils.clientside_credential_handler import ( + get_dynamic_litellm_params, + ) + + admin_params = { + "model": "bedrock/claude-3", + "aws_access_key_id": "AKIA-EXAMPLE", + "aws_secret_access_key": "secret-example", + "aws_session_token": "session-example", + "vertex_credentials": '{"private_key":"-----BEGIN..."}', + "vertex_project": "admin-gcp-project", + } + out = get_dynamic_litellm_params( + litellm_params=dict(admin_params), + request_kwargs={"base_url": "https://attacker.example"}, + ) + assert "aws_access_key_id" not in out + assert "aws_secret_access_key" not in out + assert "aws_session_token" not in out + assert "vertex_credentials" not in out + assert "vertex_project" not in out + + def test_caller_resupplied_value_overrides_admin_value_on_base_override(self): + # When the caller redirects ``api_base`` and *also* supplies their + # own value for one of the admin fields (e.g. ``organization``), + # the caller's value must win — never the admin's. The naive + # ``if field not in request_kwargs: pop`` shape lets a caller echo + # the field name with any value (or empty string) to keep the + # admin's value forwarded, which is the exfiltration vector this + # test guards against. + from litellm.router_utils.clientside_credential_handler import ( + get_dynamic_litellm_params, + ) + + out = get_dynamic_litellm_params( + litellm_params={ + "api_base": "https://admin.upstream/v1", + "organization": "org-admin", + "extra_body": {"admin": "value"}, + }, + request_kwargs={ + "api_base": "https://attacker.example", + "organization": "org-attacker", + "extra_body": {"attacker": "value"}, + }, + ) + assert out["organization"] == "org-attacker" + assert out["extra_body"] == {"attacker": "value"} + + def test_field_echo_does_not_preserve_admin_value(self): + # Regression: a caller that echoes an admin-config field name with + # an *empty* value (or any value) must not be able to keep the + # admin's value in ``litellm_params``. + from litellm.router_utils.clientside_credential_handler import ( + get_dynamic_litellm_params, + ) + + out = get_dynamic_litellm_params( + litellm_params={ + "api_base": "https://admin.upstream/v1", + "organization": "org-admin-secret", + "extra_body": {"x-admin-only": "secret"}, + }, + request_kwargs={ + "api_base": "https://attacker.example", + "organization": "", + "extra_body": "", + }, + ) + assert out["organization"] == "" + assert out["extra_body"] == "" + assert "org-admin-secret" not in str(out) + + def test_no_clearing_when_only_api_key_overridden(self): + from litellm.router_utils.clientside_credential_handler import ( + get_dynamic_litellm_params, + ) + + # Caller only overrides api_key (BYOK pattern); admin's organization / + # extra_body / region still apply because the destination is unchanged. + out = get_dynamic_litellm_params( + litellm_params={ + "api_base": "https://admin.upstream/v1", + "organization": "org-admin", + "api_version": "2026-04-01", + }, + request_kwargs={"api_key": "sk-byok"}, + ) + assert out["organization"] == "org-admin" + assert out["api_version"] == "2026-04-01" + assert out["api_base"] == "https://admin.upstream/v1" + + +class TestIsRequestBodySafeBlocksEndpointTargetingFields: + """ + ``is_request_body_safe`` rejects request-body fields that retarget the + outbound request to a caller-controlled host. Beyond the original + ``api_base`` / ``base_url``, the same protection must apply to: + + * ``aws_bedrock_runtime_endpoint`` — Bedrock endpoint redirect; an + attacker-controlled value coerces the proxy to authenticate against + their host with the admin's AWS creds. + * ``langsmith_base_url`` — Langsmith callback host; attacker-controlled + values exfiltrate the entire request payload (incl. message content) + via the observability hook. + * ``langfuse_host`` — same exfil vector via the Langfuse hook. + """ + + @pytest.fixture(autouse=True) + def _disable_url_validation(self, monkeypatch): + # The new banned-params entries should be rejected even when + # ``user_url_validation`` is off — the gate isn't the URL guard, + # it's the banned-params list. + import litellm + + monkeypatch.setattr(litellm, "user_url_validation", False, raising=False) + + @pytest.mark.parametrize( + "field", + [ + "aws_bedrock_runtime_endpoint", + "langsmith_base_url", + "langfuse_host", + "posthog_host", + "braintrust_host", + "slack_webhook_url", + "s3_endpoint_url", + "sagemaker_base_url", + "deployment_url", + ], + ) + def test_endpoint_targeting_field_in_request_body_is_rejected(self, field): + with pytest.raises(ValueError) as exc: + is_request_body_safe( + request_body={"model": "gpt-4", field: "https://attacker.example"}, + general_settings={}, + llm_router=None, + model="gpt-4", + ) + # The function lists the offending param name in the error. + assert field in str(exc.value) + + @pytest.mark.parametrize( + "field", + ["api_base", "base_url", "user_config", "langfuse_host", "slack_webhook_url"], + ) + def test_api_key_does_not_bypass_blocklist(self, field): + # Regression: the historical ``check_complete_credentials`` clause + # made the entire blocklist a no-op for any caller that supplied + # a non-empty ``api_key``. That bypass turned every missing entry + # on the blocklist into an SSRF / credential-exfil hole. Verify + # that supplying an api_key (alongside the banned param) does NOT + # bypass the gate — it can only be opened by an admin opt-in. + with pytest.raises(ValueError) as exc: + is_request_body_safe( + request_body={ + "model": "gpt-4", + "api_key": "sk-anything", + field: "https://attacker.example", + }, + general_settings={}, + llm_router=None, + model="gpt-4", + ) + assert field in str(exc.value) + + def test_admin_opt_in_proxy_wide_still_allows(self): + # ``general_settings.allow_client_side_credentials = True`` remains + # the documented proxy-wide BYOK opt-in. + assert ( + is_request_body_safe( + request_body={"model": "gpt-4", "api_base": "https://my-byok.example"}, + general_settings={"allow_client_side_credentials": True}, + llm_router=None, + model="gpt-4", + ) + is True + ) diff --git a/tests/test_litellm/proxy/db/test_prisma_self_heal.py b/tests/test_litellm/proxy/db/test_prisma_self_heal.py index af65cad09e3..3f9ba6af3af 100644 --- a/tests/test_litellm/proxy/db/test_prisma_self_heal.py +++ b/tests/test_litellm/proxy/db/test_prisma_self_heal.py @@ -35,18 +35,18 @@ async def test_attempt_db_reconnect_should_succeed(mock_proxy_logging): client = PrismaClient( database_url="mock://test", proxy_logging_obj=mock_proxy_logging ) - client.db.disconnect = AsyncMock(return_value=None) - client.db.connect = AsyncMock(return_value=None) + client.db.recreate_prisma_client = AsyncMock(return_value=None) client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) + client._start_engine_watcher = AsyncMock() - result = await client.attempt_db_reconnect( - reason="unit_test_reconnect_success", - force=True, - ) + with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}): + result = await client.attempt_db_reconnect( + reason="unit_test_reconnect_success", + force=True, + ) assert result is True - client.db.disconnect.assert_awaited_once() - client.db.connect.assert_awaited_once() + client.db.recreate_prisma_client.assert_awaited_once_with("postgresql://test") client.db.query_raw.assert_awaited_once_with("SELECT 1") @@ -141,15 +141,19 @@ async def test_attempt_db_reconnect_should_set_cooldown_after_attempt( ) client._db_last_reconnect_attempt_ts = 0.0 client._db_reconnect_cooldown_seconds = 10 - client.db.disconnect = AsyncMock(return_value=None) - client.db.connect = AsyncMock(return_value=None) + client.db.recreate_prisma_client = AsyncMock(return_value=None) client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) + client._start_engine_watcher = AsyncMock() # Use a counter-based mock to avoid StopIteration when time.time() is called # more times than expected (varies by Python version / internal code paths). fake_clock = iter(range(100, 10000)) - with patch( - "litellm.proxy.utils.time.time", side_effect=lambda: float(next(fake_clock)) + with ( + patch( + "litellm.proxy.utils.time.time", + side_effect=lambda: float(next(fake_clock)), + ), + patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}), ): result = await client.attempt_db_reconnect( reason="unit_test_cooldown_timestamp_after_attempt", @@ -163,23 +167,28 @@ async def test_attempt_db_reconnect_should_set_cooldown_after_attempt( @pytest.mark.asyncio -async def test_run_reconnect_cycle_watchdog_should_use_direct_db_ops( +async def test_run_reconnect_cycle_watchdog_should_use_recreate_prisma_client( mock_proxy_logging, ): + """Direct reconnect goes through recreate_prisma_client (which non-blockingly + kills the old engine) instead of calling disconnect() — see issue #26191. + """ client = PrismaClient( database_url="mock://test", proxy_logging_obj=mock_proxy_logging ) - client.disconnect = AsyncMock(side_effect=AssertionError("wrapper disconnect used")) - client.connect = AsyncMock(side_effect=AssertionError("wrapper connect used")) - client.db.disconnect = AsyncMock(return_value=None) - client.db.connect = AsyncMock(return_value=None) + client.db.disconnect = AsyncMock( + side_effect=AssertionError("disconnect must not be called") + ) + client.db.recreate_prisma_client = AsyncMock(return_value=None) client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) + client._start_engine_watcher = AsyncMock() - await client._run_reconnect_cycle(timeout_seconds=None) + with patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}): + await client._run_reconnect_cycle(timeout_seconds=None) - client.db.disconnect.assert_awaited_once() - client.db.connect.assert_awaited_once() + client.db.recreate_prisma_client.assert_awaited_once_with("postgresql://test") client.db.query_raw.assert_awaited_once_with("SELECT 1") + client.db.disconnect.assert_not_awaited() @pytest.mark.asyncio @@ -190,19 +199,22 @@ async def test_run_reconnect_cycle_watchdog_should_use_default_timeout_budget( database_url="mock://test", proxy_logging_obj=mock_proxy_logging ) client._db_watchdog_reconnect_timeout_seconds = 0.1 - client.db.disconnect = AsyncMock(return_value=None) + client._start_engine_watcher = AsyncMock() - async def _slow_connect(): + async def _slow_recreate(_db_url): await asyncio.sleep(0.08) async def _slow_query(_query: str): await asyncio.sleep(0.08) return [{"result": 1}] - client.db.connect = AsyncMock(side_effect=_slow_connect) + client.db.recreate_prisma_client = AsyncMock(side_effect=_slow_recreate) client.db.query_raw = AsyncMock(side_effect=_slow_query) - with pytest.raises(asyncio.TimeoutError): + with ( + pytest.raises(asyncio.TimeoutError), + patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}), + ): await client._run_reconnect_cycle(timeout_seconds=None) @@ -213,19 +225,22 @@ async def test_run_reconnect_cycle_timeout_should_use_single_overall_budget( client = PrismaClient( database_url="mock://test", proxy_logging_obj=mock_proxy_logging ) - client.db.disconnect = AsyncMock(return_value=None) + client._start_engine_watcher = AsyncMock() - async def _slow_connect(): + async def _slow_recreate(_db_url): await asyncio.sleep(0.08) async def _slow_query(_query: str): await asyncio.sleep(0.08) return [{"result": 1}] - client.db.connect = AsyncMock(side_effect=_slow_connect) + client.db.recreate_prisma_client = AsyncMock(side_effect=_slow_recreate) client.db.query_raw = AsyncMock(side_effect=_slow_query) - with pytest.raises(asyncio.TimeoutError): + with ( + pytest.raises(asyncio.TimeoutError), + patch.dict(os.environ, {"DATABASE_URL": "postgresql://test"}), + ): await client._run_reconnect_cycle(timeout_seconds=0.1) @@ -320,45 +335,35 @@ async def test_db_health_watchdog_start_stop_lifecycle(mock_proxy_logging): @pytest.mark.asyncio -async def test_lightweight_reconnect_kills_engine_on_disconnect_failure( +async def test_recreate_prisma_client_kills_old_engine_without_disconnect( mock_proxy_logging, ): - """Lightweight reconnect must kill the old engine PID when disconnect() fails.""" + """recreate_prisma_client SIGTERMs the old engine PID directly rather than + calling `disconnect()`, which blocks the asyncio event loop on the sync + `subprocess.Popen.wait()` inside prisma-client-py — see issue #26191. + """ client = PrismaClient( database_url="mock://test", proxy_logging_obj=mock_proxy_logging ) - client.db.disconnect = AsyncMock(side_effect=Exception("disconnect failed")) - client.db.connect = AsyncMock(return_value=None) - client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) + disconnect_mock = AsyncMock( + side_effect=AssertionError("disconnect must not be called on reconnect path") + ) + client.db._original_prisma.disconnect = disconnect_mock with ( - patch.object(client, "_get_engine_pid", return_value=9999), - patch("os.kill") as mock_kill, - patch("asyncio.sleep", new_callable=AsyncMock), + patch.object(client.db, "_get_engine_pid", return_value=9999), + patch("litellm.proxy.db.prisma_client.os.kill") as mock_kill, + patch("litellm.proxy.db.prisma_client.asyncio.sleep", new_callable=AsyncMock), ): - await client._run_reconnect_cycle(timeout_seconds=5.0) + # Return a Prisma instance whose connect() is awaitable. + fake_new_prisma = MagicMock() + fake_new_prisma.connect = AsyncMock(return_value=None) + with patch("prisma.Prisma", return_value=fake_new_prisma): + await client.db.recreate_prisma_client("postgresql://test") mock_kill.assert_any_call(9999, signal.SIGTERM) - client.db.connect.assert_awaited_once() - client.db.query_raw.assert_awaited_once_with("SELECT 1") - - -@pytest.mark.asyncio -async def test_lightweight_reconnect_skips_kill_on_successful_disconnect( - mock_proxy_logging, -): - """Lightweight reconnect must NOT kill when disconnect() succeeds.""" - client = PrismaClient( - database_url="mock://test", proxy_logging_obj=mock_proxy_logging - ) - client.db.disconnect = AsyncMock(return_value=None) - client.db.connect = AsyncMock(return_value=None) - client.db.query_raw = AsyncMock(return_value=[{"result": 1}]) - - with patch("os.kill") as mock_kill: - await client._run_reconnect_cycle(timeout_seconds=5.0) - - mock_kill.assert_not_called() + disconnect_mock.assert_not_awaited() + fake_new_prisma.connect.assert_awaited_once() # --------------------------------------------------------------------------- diff --git a/tests/test_litellm/proxy/management_endpoints/test_workflow_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_workflow_management_endpoints.py new file mode 100644 index 00000000000..a337ff6d888 --- /dev/null +++ b/tests/test_litellm/proxy/management_endpoints/test_workflow_management_endpoints.py @@ -0,0 +1,611 @@ +""" +Unit tests for workflow management endpoints (/v1/workflows/runs/*). +Uses FastAPI TestClient with a mocked prisma_client. +""" + +import os +import sys +from datetime import datetime, timezone +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi import FastAPI +from fastapi.testclient import TestClient +from prisma.errors import UniqueViolationError + +sys.path.insert(0, os.path.abspath("../../..")) + +from litellm.proxy.management_endpoints.workflow_management_endpoints import router + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_run( + run_id: str = "run-1", + session_id: str = "sess-1", + workflow_type: str = "shin-builder", + status: str = "pending", + created_by: Any = "tok-test", +) -> MagicMock: + obj = MagicMock() + obj.run_id = run_id + obj.session_id = session_id + obj.workflow_type = workflow_type + obj.status = status + obj.created_by = created_by + obj.created_at = datetime.now(timezone.utc) + obj.updated_at = datetime.now(timezone.utc) + obj.input = None + obj.output = None + obj.metadata = None + return obj + + +def _make_event( + event_id: str = "evt-1", + run_id: str = "run-1", + event_type: str = "step.started", + step_name: str = "grill", + sequence_number: int = 0, +) -> MagicMock: + obj = MagicMock() + obj.event_id = event_id + obj.run_id = run_id + obj.event_type = event_type + obj.step_name = step_name + obj.sequence_number = sequence_number + obj.data = None + obj.created_at = datetime.now(timezone.utc) + return obj + + +def _make_message( + message_id: str = "msg-1", + run_id: str = "run-1", + role: str = "user", + content: str = "hello", + sequence_number: int = 0, +) -> MagicMock: + obj = MagicMock() + obj.message_id = message_id + obj.run_id = run_id + obj.role = role + obj.content = content + obj.sequence_number = sequence_number + obj.session_id = None + obj.created_at = datetime.now(timezone.utc) + return obj + + +def _make_tx(event_return=None, run_return=None, msg_return=None) -> MagicMock: + """Build an async context-manager mock for prisma_client.db.tx().""" + tx = MagicMock() + tx.litellm_workflowevent = MagicMock() + tx.litellm_workflowevent.create = AsyncMock( + return_value=event_return or _make_event() + ) + tx.litellm_workflowrun = MagicMock() + tx.litellm_workflowrun.update = AsyncMock(return_value=run_return or _make_run()) + tx.litellm_workflowmessage = MagicMock() + tx.litellm_workflowmessage.create = AsyncMock( + return_value=msg_return or _make_message() + ) + tx.__aenter__ = AsyncMock(return_value=tx) + tx.__aexit__ = AsyncMock(return_value=False) + return tx + + +def _make_prisma_client() -> MagicMock: + client = MagicMock() + client.db = MagicMock() + client.db.litellm_workflowrun = MagicMock() + client.db.litellm_workflowevent = MagicMock() + client.db.litellm_workflowmessage = MagicMock() + # default tx() returns a no-op transaction + client.db.tx = MagicMock(return_value=_make_tx()) + return client + + +def _make_app() -> FastAPI: + app = FastAPI() + app.include_router(router) + return app + + +def _override_auth() -> Any: + from litellm.proxy._types import UserAPIKeyAuth + + auth = UserAPIKeyAuth(api_key="sk-test", user_id="admin") + auth.token = "tok-test" + return auth + + +def _override_auth_admin() -> Any: + from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth + + auth = UserAPIKeyAuth(api_key="sk-master") + auth.user_role = LitellmUserRoles.PROXY_ADMIN # type: ignore[assignment] + return auth + + +def _override_auth_user_with_token(token: str = "tok-abc") -> Any: + """Return a non-admin caller whose hashed token equals `token`.""" + from litellm.proxy._types import UserAPIKeyAuth + + auth = UserAPIKeyAuth(api_key="sk-user", user_id="user-1") + auth.token = token # override the computed hash with a predictable value + return auth + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestCreateWorkflowRun: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_create_returns_run(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.create = AsyncMock(return_value=_make_run()) + + resp = self.client.post( + "/v1/workflows/runs", + json={"workflow_type": "shin-builder"}, + ) + assert resp.status_code == 200 + self._prisma.db.litellm_workflowrun.create.assert_awaited_once() + + @patch("litellm.proxy.proxy_server.prisma_client", None) + def test_create_500_when_no_db(self): + resp = self.client.post( + "/v1/workflows/runs", + json={"workflow_type": "shin-builder"}, + ) + assert resp.status_code == 500 + + +class TestListWorkflowRuns: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_returns_runs(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_many = AsyncMock( + return_value=[_make_run()] + ) + + resp = self.client.get("/v1/workflows/runs") + assert resp.status_code == 200 + data = resp.json() + assert data["count"] == 1 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_filters_by_status(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_many = AsyncMock(return_value=[]) + + resp = self.client.get("/v1/workflows/runs?status=running") + assert resp.status_code == 200 + call_kwargs = self._prisma.db.litellm_workflowrun.find_many.call_args[1] + assert call_kwargs["where"]["status"] == "running" + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_filters_by_multiple_statuses(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_many = AsyncMock(return_value=[]) + + resp = self.client.get("/v1/workflows/runs?status=running,paused") + assert resp.status_code == 200 + call_kwargs = self._prisma.db.litellm_workflowrun.find_many.call_args[1] + assert call_kwargs["where"]["status"] == {"in": ["running", "paused"]} + + +class TestGetWorkflowRun: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_get_existing_run(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + + resp = self.client.get("/v1/workflows/runs/run-1") + assert resp.status_code == 200 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_get_missing_run_returns_404(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock(return_value=None) + + resp = self.client.get("/v1/workflows/runs/nonexistent") + assert resp.status_code == 404 + + +class TestUpdateWorkflowRun: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_update_status(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + updated = _make_run(status="completed") + self._prisma.db.litellm_workflowrun.update = AsyncMock(return_value=updated) + + resp = self.client.patch( + "/v1/workflows/runs/run-1", json={"status": "completed"} + ) + assert resp.status_code == 200 + self._prisma.db.litellm_workflowrun.update.assert_awaited_once() + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_update_no_fields_returns_400(self, mock_pc): + mock_pc.db = self._prisma.db + resp = self.client.patch("/v1/workflows/runs/run-1", json={}) + assert resp.status_code == 400 + + +class TestAppendWorkflowEvent: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_append_event_updates_run_status(self, mock_pc): + mock_pc.db = self._prisma.db + # _require_run check + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowevent.find_many = AsyncMock(return_value=[]) + tx = _make_tx( + event_return=_make_event(), run_return=_make_run(status="running") + ) + self._prisma.db.tx = MagicMock(return_value=tx) + + resp = self.client.post( + "/v1/workflows/runs/run-1/events", + json={"event_type": "step.started", "step_name": "grill"}, + ) + assert resp.status_code == 200 + # run status updated inside tx + tx.litellm_workflowrun.update.assert_awaited_once() + update_call = tx.litellm_workflowrun.update.call_args[1] + assert update_call["data"]["status"] == "running" + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_append_event_no_status_update_for_unknown_type(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowevent.find_many = AsyncMock(return_value=[]) + tx = _make_tx(event_return=_make_event(event_type="custom.event")) + self._prisma.db.tx = MagicMock(return_value=tx) + + resp = self.client.post( + "/v1/workflows/runs/run-1/events", + json={"event_type": "custom.event", "step_name": "grill"}, + ) + assert resp.status_code == 200 + # no status update inside tx for unknown event_type + tx.litellm_workflowrun.update.assert_not_awaited() + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_sequence_number_increments(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + existing = _make_event(sequence_number=4) + self._prisma.db.litellm_workflowevent.find_many = AsyncMock( + return_value=[existing] + ) + tx = _make_tx(event_return=_make_event(sequence_number=5)) + self._prisma.db.tx = MagicMock(return_value=tx) + + self.client.post( + "/v1/workflows/runs/run-1/events", + json={"event_type": "step.started", "step_name": "plan"}, + ) + create_call = tx.litellm_workflowevent.create.call_args[1] + assert create_call["data"]["sequence_number"] == 5 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_unknown_run_id_returns_404(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock(return_value=None) + + resp = self.client.post( + "/v1/workflows/runs/nonexistent/events", + json={"event_type": "step.started", "step_name": "grill"}, + ) + assert resp.status_code == 404 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_sequence_collision_retries_and_succeeds(self, mock_pc): + """UniqueViolationError on first attempt triggers retry; second attempt succeeds.""" + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowevent.find_many = AsyncMock(return_value=[]) + + # First tx raises UniqueViolationError; second succeeds. + tx_fail = _make_tx() + tx_fail.__aenter__ = AsyncMock(return_value=tx_fail) + tx_fail.litellm_workflowevent.create = AsyncMock( + side_effect=UniqueViolationError( + {"user_facing_error": {"message": "unique"}} + ) + ) + tx_fail.__aexit__ = AsyncMock(return_value=False) + + tx_ok = _make_tx(event_return=_make_event(sequence_number=1)) + + self._prisma.db.tx = MagicMock(side_effect=[tx_fail, tx_ok]) + + resp = self.client.post( + "/v1/workflows/runs/run-1/events", + json={"event_type": "step.started", "step_name": "grill"}, + ) + assert resp.status_code == 200 + + +class TestWorkflowMessages: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_append_message(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowmessage.find_many = AsyncMock(return_value=[]) + self._prisma.db.litellm_workflowmessage.create = AsyncMock( + return_value=_make_message() + ) + + resp = self.client.post( + "/v1/workflows/runs/run-1/messages", + json={"role": "user", "content": "fix the bug"}, + ) + assert resp.status_code == 200 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_append_message_unknown_run_returns_404(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock(return_value=None) + + resp = self.client.post( + "/v1/workflows/runs/nonexistent/messages", + json={"role": "user", "content": "hello"}, + ) + assert resp.status_code == 404 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_messages_ordered(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowmessage.find_many = AsyncMock( + return_value=[ + _make_message(sequence_number=0), + _make_message(sequence_number=1, role="assistant"), + ] + ) + + resp = self.client.get("/v1/workflows/runs/run-1/messages") + assert resp.status_code == 200 + data = resp.json() + assert data["count"] == 2 + call_kwargs = self._prisma.db.litellm_workflowmessage.find_many.call_args[1] + assert call_kwargs["order"] == {"sequence_number": "asc"} + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_messages_respects_limit(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowmessage.find_many = AsyncMock(return_value=[]) + + resp = self.client.get("/v1/workflows/runs/run-1/messages?limit=25") + assert resp.status_code == 200 + call_kwargs = self._prisma.db.litellm_workflowmessage.find_many.call_args[1] + assert call_kwargs["take"] == 25 + + +class TestListWorkflowEvents: + def setup_method(self): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = _override_auth + self.client = TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_events_ordered(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowevent.find_many = AsyncMock( + return_value=[ + _make_event(sequence_number=0), + _make_event(sequence_number=1), + ] + ) + + resp = self.client.get("/v1/workflows/runs/run-1/events") + assert resp.status_code == 200 + data = resp.json() + assert data["count"] == 2 + call_kwargs = self._prisma.db.litellm_workflowevent.find_many.call_args[1] + assert call_kwargs["order"] == {"sequence_number": "asc"} + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_events_respects_limit(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run() + ) + self._prisma.db.litellm_workflowevent.find_many = AsyncMock(return_value=[]) + + resp = self.client.get("/v1/workflows/runs/run-1/events?limit=10") + assert resp.status_code == 200 + call_kwargs = self._prisma.db.litellm_workflowevent.find_many.call_args[1] + assert call_kwargs["take"] == 10 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_list_events_unknown_run_returns_404(self, mock_pc): + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock(return_value=None) + + resp = self.client.get("/v1/workflows/runs/nonexistent/events") + assert resp.status_code == 404 + + +class TestTenantIsolation: + """Ownership enforcement: non-admin callers only see their own runs.""" + + def _make_app_with_auth(self, auth_fn): + from litellm.proxy.auth.user_api_key_auth import user_api_key_auth + + self._prisma = _make_prisma_client() + app = _make_app() + app.dependency_overrides[user_api_key_auth] = auth_fn + return TestClient(app, raise_server_exceptions=True) + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_create_stores_caller_token(self, mock_pc): + token = "tok-owner" + client = self._make_app_with_auth(lambda: _override_auth_user_with_token(token)) + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.create = AsyncMock( + return_value=_make_run(created_by=token) + ) + + resp = client.post("/v1/workflows/runs", json={"workflow_type": "test"}) + assert resp.status_code == 200 + create_call = self._prisma.db.litellm_workflowrun.create.call_args[1] + assert create_call["data"]["created_by"] == token + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_non_admin_list_scoped_to_caller_token(self, mock_pc): + token = "tok-owner" + client = self._make_app_with_auth(lambda: _override_auth_user_with_token(token)) + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_many = AsyncMock(return_value=[]) + + resp = client.get("/v1/workflows/runs") + assert resp.status_code == 200 + call_kwargs = self._prisma.db.litellm_workflowrun.find_many.call_args[1] + assert call_kwargs["where"].get("created_by") == token + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_admin_list_not_scoped(self, mock_pc): + client = self._make_app_with_auth(_override_auth_admin) + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_many = AsyncMock(return_value=[]) + + resp = client.get("/v1/workflows/runs") + assert resp.status_code == 200 + call_kwargs = self._prisma.db.litellm_workflowrun.find_many.call_args[1] + assert "created_by" not in call_kwargs["where"] + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_non_admin_get_other_users_run_returns_404(self, mock_pc): + token = "tok-caller" + client = self._make_app_with_auth(lambda: _override_auth_user_with_token(token)) + mock_pc.db = self._prisma.db + # Run owned by a different key + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run(created_by="tok-other-owner") + ) + + resp = client.get("/v1/workflows/runs/run-1") + assert resp.status_code == 404 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_non_admin_get_null_owner_run_returns_404(self, mock_pc): + token = "tok-caller" + client = self._make_app_with_auth(lambda: _override_auth_user_with_token(token)) + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run(created_by=None) + ) + + resp = client.get("/v1/workflows/runs/run-1") + assert resp.status_code == 404 + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_non_admin_update_null_owner_run_returns_404(self, mock_pc): + token = "tok-caller" + client = self._make_app_with_auth(lambda: _override_auth_user_with_token(token)) + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run(created_by=None) + ) + self._prisma.db.litellm_workflowrun.update = AsyncMock( + return_value=_make_run(status="completed") + ) + + resp = client.patch("/v1/workflows/runs/run-1", json={"status": "completed"}) + assert resp.status_code == 404 + self._prisma.db.litellm_workflowrun.update.assert_not_awaited() + + @patch("litellm.proxy.proxy_server.prisma_client") + def test_non_admin_get_own_run_succeeds(self, mock_pc): + token = "tok-caller" + client = self._make_app_with_auth(lambda: _override_auth_user_with_token(token)) + mock_pc.db = self._prisma.db + self._prisma.db.litellm_workflowrun.find_unique = AsyncMock( + return_value=_make_run(created_by=token) + ) + + resp = client.get("/v1/workflows/runs/run-1") + assert resp.status_code == 200 diff --git a/ui/litellm-dashboard/src/app/page.tsx b/ui/litellm-dashboard/src/app/page.tsx index d32cf74d6ad..a8553d5405b 100644 --- a/ui/litellm-dashboard/src/app/page.tsx +++ b/ui/litellm-dashboard/src/app/page.tsx @@ -40,6 +40,7 @@ import { ProjectsPage } from "@/components/Projects/ProjectsPage"; import VectorStoreManagement from "@/components/vector_store_management"; import ToolPoliciesView from "@/components/ToolPoliciesView"; import { MemoryView } from "@/components/MemoryView"; +import WorkflowRuns from "@/components/workflow_runs"; import SpendLogsTable from "@/components/view_logs"; import ViewUserDashboard from "@/components/view_users"; import { ThemeProvider } from "@/contexts/ThemeContext"; @@ -631,6 +632,8 @@ function CreateKeyPageContent() { ) : page == "tool-policies" ? ( + ) : page == "workflows" ? ( + ) : page == "memory" ? ( { "Virtual Keys", "Playground", "Models + Endpoints", - "Agents", + "Agentic", "MCP Servers", "Guardrails", "Policies", @@ -163,10 +163,17 @@ describe("Sidebar (leftnav)", () => { expect(screen.getByText("Models + Endpoints")).toBeInTheDocument(); }); - it("shows Agents to Admin Viewer (read-only)", () => { + it("shows Agents (under Agentic) to Admin Viewer (read-only)", async () => { mockUseAuthorized.mockReturnValueOnce(adminViewerAuth); renderWithProviders(); - expect(screen.getByText("Agents")).toBeInTheDocument(); + // Agents is now nested under the "Agentic" submenu — expand parent + // first to render the children, then assert Agents is visible. + act(() => { + fireEvent.click(screen.getByText("Agentic")); + }); + await waitFor(() => { + expect(screen.getByText("Agents")).toBeInTheDocument(); + }); }); it("shows Logs to Admin Viewer", () => { diff --git a/ui/litellm-dashboard/src/components/leftnav.tsx b/ui/litellm-dashboard/src/components/leftnav.tsx index 029bdc49b5d..f2fe1fe96ec 100644 --- a/ui/litellm-dashboard/src/components/leftnav.tsx +++ b/ui/litellm-dashboard/src/components/leftnav.tsx @@ -3,6 +3,7 @@ import { useTeams } from "@/app/(dashboard)/hooks/teams/useTeams"; import useAuthorized from "@/app/(dashboard)/hooks/useAuthorized"; import { ApiOutlined, + ApartmentOutlined, AppstoreOutlined, AuditOutlined, BankOutlined, @@ -129,11 +130,33 @@ const menuGroups: MenuGroup[] = [ roles: rolesAllowedToViewWriteScopedPages, }, { - key: "agents", - page: "agents", - label: "Agents", + key: "agentic", + page: "agentic", + label: "Agentic", icon: , - roles: rolesAllowedToViewWriteScopedPages, + children: [ + { + key: "agents", + page: "agents", + label: "Agents", + icon: , + // Admin Viewer can view agents read-only (write actions are + // hidden inside the page); Playground above stays write-only. + roles: rolesAllowedToViewWriteScopedPages, + }, + { + key: "workflows", + page: "workflows", + label: "Workflow Runs", + icon: , + }, + { + key: "memory", + page: "memory", + label: "Memory", + icon: , + }, + ], }, { key: "mcp-servers", @@ -148,12 +171,6 @@ const menuGroups: MenuGroup[] = [ icon: , roles: all_admin_roles, }, - { - key: "memory", - page: "memory", - label: "Memory", - icon: , - }, { key: "guardrails", page: "guardrails", diff --git a/ui/litellm-dashboard/src/components/page_metadata.ts b/ui/litellm-dashboard/src/components/page_metadata.ts index 63f03f4586c..277bb0bd021 100644 --- a/ui/litellm-dashboard/src/components/page_metadata.ts +++ b/ui/litellm-dashboard/src/components/page_metadata.ts @@ -9,6 +9,8 @@ export const pageDescriptions: Record = { "llm-playground": "Interactive playground for testing LLM requests", models: "Configure and manage LLM models and endpoints", agents: "Create and manage AI agents", + agentic: "Manage agentic resources: agents, workflow runs, and memory", + workflows: "Track and inspect durable workflow run history", "mcp-servers": "Configure Model Context Protocol servers", memory: "Inspect and manage agent memory entries stored under /v1/memory", guardrails: "Set up content moderation and safety guardrails", diff --git a/ui/litellm-dashboard/src/components/workflow_runs/index.tsx b/ui/litellm-dashboard/src/components/workflow_runs/index.tsx new file mode 100644 index 00000000000..a69eac282bf --- /dev/null +++ b/ui/litellm-dashboard/src/components/workflow_runs/index.tsx @@ -0,0 +1,751 @@ +import React, { useState, useEffect, useCallback } from "react"; +import { Button, Collapse, Drawer, Empty, Spin, Table, Tooltip, Typography } from "antd"; +import { ReloadOutlined } from "@ant-design/icons"; +import { proxyBaseUrl } from "@/components/networking"; + +const { Text } = Typography; + +interface WorkflowRunsProps { + accessToken: string | null; +} + +type RunStatus = "pending" | "running" | "paused" | "completed" | "failed"; + +interface RunMetadata { + title?: string; + state?: string; + pr_url?: string; + worktree_path?: string; + plan_text?: string; + grill_session_id?: string; + session_id?: string; + [key: string]: unknown; +} + +interface WorkflowRun { + run_id: string; + status: RunStatus; + workflow_type: string; + created_at: string; + metadata?: RunMetadata | null; +} + +interface WorkflowRunEvent { + event_id: string; + event_type: string; + step_name: string; + sequence_number: number; + created_at: string; + data?: Record | null; +} + +interface WorkflowRunMessage { + message_id: string; + role: string; + content: string; + sequence_number: number; + created_at: string; +} + +// ── design tokens ───────────────────────────────────────────────────────────── + +const STATUS_DOT: Record = { + pending: "#a1a1aa", + running: "#3b82f6", + paused: "#f59e0b", + completed: "#22c55e", + failed: "#ef4444", +}; + +const EVENT_COLOR: Record = { + "step.started": { bar: "#f0fdf4", border: "#86efac", text: "#16a34a" }, + "step.failed": { bar: "#fef2f2", border: "#fca5a5", text: "#dc2626" }, + "hook.waiting": { bar: "#fffbeb", border: "#fcd34d", text: "#d97706" }, + "hook.received": { bar: "#eff6ff", border: "#93c5fd", text: "#2563eb" }, +}; + +function eventStyle(type: string) { + return EVENT_COLOR[type] ?? { bar: "#f4f4f5", border: "#d4d4d8", text: "#52525b" }; +} + +// ── helpers ─────────────────────────────────────────────────────────────────── + +function timeAgo(iso: string): string { + const diff = Date.now() - new Date(iso).getTime(); + if (isNaN(diff)) return iso; + const s = Math.floor(diff / 1000); + if (s < 60) return `${s}s ago`; + const m = Math.floor(s / 60); + if (m < 60) return `${m}m ago`; + const h = Math.floor(m / 60); + if (h < 24) return `${h}h ago`; + return `${Math.floor(h / 24)}d ago`; +} + +function fmtDuration(ms: number): string { + if (ms < 0) return ""; + if (ms < 1000) return `${ms}ms`; + return `${(ms / 1000).toFixed(1)}s`; +} + +function runTitle(run: WorkflowRun): string { + const t = run.metadata?.title; + if (t) return String(t); + return run.workflow_type ?? run.run_id.slice(0, 8); +} + +function shortId(id: string): string { + return id.slice(0, 8); +} + +// ── status dot ──────────────────────────────────────────────────────────────── + +const StatusDot: React.FC<{ status: RunStatus; size?: number }> = ({ status, size = 8 }) => ( + +); + +// ── truncated text value ────────────────────────────────────────────────────── + +const TRUNCATE_AT = 120; + +const TruncatedValue: React.FC<{ value: string }> = ({ value }) => { + const [expanded, setExpanded] = useState(false); + if (value.length <= TRUNCATE_AT) { + return {value}; + } + return ( + + {expanded ? value : value.slice(0, TRUNCATE_AT) + "…"} + + + ); +}; + +// ── metadata card ───────────────────────────────────────────────────────────── + +const MetadataCard: React.FC<{ run: WorkflowRun }> = ({ run }) => { + const meta = run.metadata ?? {}; + + const primaryFields: { key: string; label: string }[] = [ + { key: "state", label: "state" }, + { key: "worktree_path", label: "worktree" }, + { key: "grill_session_id", label: "grill session" }, + { key: "session_id", label: "session" }, + ]; + + const primaryKeys = new Set(["title", ...primaryFields.map((f) => f.key)]); + const extraEntries = Object.entries(meta).filter( + ([k, v]) => !primaryKeys.has(k) && v !== null && v !== undefined && v !== "" + ); + + return ( +
+ {/* title bar */} +
+ + + {runTitle(run)} + + + {shortId(run.run_id)} + + + {run.workflow_type} + +
+ + {/* key fields grid */} +
+ + {run.status} + + + {timeAgo(run.created_at)} + + + {meta.pr_url && ( + + + {String(meta.pr_url)} + + + )} + + {primaryFields.map(({ key, label }) => { + const v = meta[key]; + if (v === null || v === undefined || v === "") return null; + const str = typeof v === "object" ? JSON.stringify(v) : String(v); + return ( + + + + ); + })} + + {extraEntries.map(([k, v]) => { + const str = typeof v === "object" ? JSON.stringify(v) : String(v); + return ( + + + + ); + })} +
+
+ ); +}; + +const FieldPair: React.FC<{ label: string; children: React.ReactNode }> = ({ + label, + children, +}) => ( +
+ + {label} + + {children} +
+); + +// ── gantt timeline ──────────────────────────────────────────────────────────── + +const GanttTimeline: React.FC<{ + run: WorkflowRun; + events: WorkflowRunEvent[]; +}> = ({ run, events }) => { + if (events.length === 0) { + return ( +
+ No events recorded +
+ ); + } + + const runStart = new Date(run.created_at).getTime(); + const eventTimes = events.map((e) => new Date(e.created_at).getTime()); + const lastTime = Math.max(...eventTimes); + const totalSpan = Math.max(lastTime - runStart, 1); + const totalDur = fmtDuration(lastTime - runStart); + + return ( +
+ {/* ruler */} +
+
+
+ {[0, 100].map((pct) => ( + + {pct === 0 ? "0" : totalDur} + + ))} +
+
+ + {/* outer run bar */} +
+
+ {runTitle(run)} +
+
+ {totalDur} +
+
+ + {/* event rows */} +
+ {events.map((ev) => { + const evTime = new Date(ev.created_at).getTime(); + const leftPct = ((evTime - runStart) / totalSpan) * 100; + + const nextIdx = events.findIndex((e) => e.sequence_number > ev.sequence_number); + const nextTime = + nextIdx >= 0 + ? new Date(events[nextIdx].created_at).getTime() + : lastTime + Math.max(totalSpan * 0.12, 500); + const widthPct = Math.max(8, ((nextTime - evTime) / totalSpan) * 100); + const style = eventStyle(ev.event_type); + const dur = fmtDuration(nextTime - evTime); + + return ( + +
+ {ev.step_name || ev.event_type} +
+
+ +
type: {ev.event_type}
+
step: {ev.step_name}
+
seq: {ev.sequence_number}
+
time: {timeAgo(ev.created_at)}
+ {ev.data && Object.keys(ev.data).length > 0 && ( +
data: {JSON.stringify(ev.data)}
+ )} +
+ } + > +
+ {ev.event_type} + {dur && {dur}} +
+ +
+ + ); + })} +
+
+ ); +}; + +// ── message row ─────────────────────────────────────────────────────────────── + +const MessageRow: React.FC<{ msg: WorkflowRunMessage }> = ({ msg }) => { + const roleColor: Record = { + user: "#2563eb", + assistant: "#16a34a", + system: "#7c3aed", + tool_result: "#d97706", + }; + const color = roleColor[msg.role] ?? "#52525b"; + + return ( +
+ [{msg.role}] +
+ + {msg.content} + + + {timeAgo(msg.created_at)} + +
+
+ ); +}; + +// ── main component ──────────────────────────────────────────────────────────── + +const WorkflowRuns: React.FC = ({ accessToken }) => { + const [runs, setRuns] = useState([]); + const [loadingRuns, setLoadingRuns] = useState(false); + const [selectedRun, setSelectedRun] = useState(null); + const [events, setEvents] = useState([]); + const [messages, setMessages] = useState([]); + const [loadingDetail, setLoadingDetail] = useState(false); + const [drawerOpen, setDrawerOpen] = useState(false); + + const fetchRuns = useCallback(async () => { + if (!accessToken) return; + setLoadingRuns(true); + try { + const res = await fetch(`${proxyBaseUrl ?? ""}/v1/workflows/runs?limit=100`, { + headers: { Authorization: `Bearer ${accessToken}` }, + }); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + const data = await res.json(); + setRuns(data.runs ?? []); + } catch (err) { + console.error("workflow runs fetch failed:", err); + } finally { + setLoadingRuns(false); + } + }, [accessToken]); + + const fetchRunDetail = useCallback( + async (run: WorkflowRun) => { + if (!accessToken) return; + setSelectedRun(run); + setDrawerOpen(true); + setLoadingDetail(true); + setEvents([]); + setMessages([]); + try { + const base = proxyBaseUrl ?? ""; + const [evRes, msgRes] = await Promise.all([ + fetch(`${base}/v1/workflows/runs/${run.run_id}/events`, { + headers: { Authorization: `Bearer ${accessToken}` }, + }), + fetch(`${base}/v1/workflows/runs/${run.run_id}/messages`, { + headers: { Authorization: `Bearer ${accessToken}` }, + }), + ]); + const evData = evRes.ok ? await evRes.json() : { events: [] }; + const msgData = msgRes.ok ? await msgRes.json() : { messages: [] }; + setEvents( + [...(evData.events ?? [])].sort( + (a: WorkflowRunEvent, b: WorkflowRunEvent) => a.sequence_number - b.sequence_number + ) + ); + setMessages( + [...(msgData.messages ?? [])].sort( + (a: WorkflowRunMessage, b: WorkflowRunMessage) => a.sequence_number - b.sequence_number + ) + ); + } catch (err) { + console.error("workflow run detail fetch failed:", err); + } finally { + setLoadingDetail(false); + } + }, + [accessToken] + ); + + useEffect(() => { + fetchRuns(); + }, [fetchRuns]); + + const columns = [ + { + title: "Run", + dataIndex: "run_id", + key: "run", + render: (_: string, run: WorkflowRun) => ( +
+ +
+
+ {runTitle(run)} +
+
+ {shortId(run.run_id)} +
+
+
+ ), + }, + { + title: "Type", + dataIndex: "workflow_type", + key: "workflow_type", + render: (v: string) => ( + {v} + ), + }, + { + title: "Status", + dataIndex: "status", + key: "status", + render: (status: RunStatus, run: WorkflowRun) => { + const state = run.metadata?.state; + return ( +
+ + + {state ?? status} + +
+ ); + }, + }, + { + title: "Created", + dataIndex: "created_at", + key: "created_at", + render: (v: string) => ( + {timeAgo(v)} + ), + }, + ]; + + return ( +
+ {/* page header */} +
+
+
Workflow Runs
+
+ Durable state tracking for agents and automated workflows +
+
+ +
+ + {/* runs table — matches logs page density */} +
+ ({ + onClick: () => fetchRunDetail(run), + style: { cursor: "pointer" }, + })} + locale={{ + emptyText: ( + No workflow runs yet} + image={Empty.PRESENTED_IMAGE_SIMPLE} + /> + ), + }} + className="[&_.ant-table-cell]:py-0.5 [&_.ant-table-thead_.ant-table-cell]:py-1" + style={{ border: "none" }} + /> + + + {/* detail drawer */} + setDrawerOpen(false)} + width={680} + title={null} + closable={false} + bodyStyle={{ padding: 0 }} + styles={{ body: { padding: 0 } }} + > + {!selectedRun ? null : loadingDetail ? ( +
+ +
+ ) : ( +
+ {/* drawer close + refresh */} +
+ + +
+ + {/* metadata card — top */} + + + {/* collapsible sections */} + + Timeline + + {events.length} {events.length === 1 ? "event" : "events"} + + + ), + children: ( +
+ +
+ ), + }, + { + key: "messages", + label: ( + + Messages + + {messages.length} + + + ), + children: messages.length === 0 ? ( +
+ No messages +
+ ) : ( +
+ {messages.map((msg) => ( + + ))} +
+ ), + }, + ]} + /> +
+ )} + + + ); +}; + +export default WorkflowRuns;