diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index 1373db220d6..4cf6db6f257 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -34909,7 +34909,6 @@ }, "name": { "default": "Lens worker", - "maxLength": 100, "minLength": 1, "title": "Name", "type": "string" diff --git a/litellm/proxy/lens/analysis.py b/litellm/proxy/lens/analysis.py index 001489f3123..7286cec9ba8 100644 --- a/litellm/proxy/lens/analysis.py +++ b/litellm/proxy/lens/analysis.py @@ -7,7 +7,7 @@ from itertools import chain, islice from types import MappingProxyType from typing import Final, Literal, TypeAlias, TypeVar -from pydantic import Field, ValidationError +from pydantic import Field, TypeAdapter, ValidationError from .models import ( Claim, @@ -31,8 +31,8 @@ from .trace_store import TraceStore, overview_content, trace_store class Observation(Record): check_id: str kind: Literal["issue", "pattern"] = "issue" - summary: str = Field(max_length=2000) - evidence: tuple[Evidence, ...] = Field(default=(), max_length=6) + summary: str + evidence: tuple[Evidence, ...] = Field(default=()) class Extraction(Record): @@ -47,14 +47,14 @@ class SpanRead(Record): class TraceReview(Extraction): feedback_page: int | None = Field(default=None, ge=0) - reads: tuple[SpanRead, ...] = Field(default=(), max_length=2) + reads: tuple[SpanRead, ...] = Field(default=()) class Candidate(Record): check_id: str kind: Literal["issue", "pattern"] = "issue" - title: str = Field(max_length=160) - hypothesis: str = Field(max_length=2000) + title: str + hypothesis: str execution_ids: tuple[str, ...] existing_finding_id: str | None = None @@ -64,7 +64,7 @@ class Clusters(Record): class Decision(Record): - action: Literal["read", "observations", "catalog", "feedback", "submit", "inconclusive"] + action: Literal["read", "evidence", "observations", "catalog", "feedback", "submit", "inconclusive"] page: int = Field(default=0, ge=0) execution_id: str | None = None cursor: str = "" @@ -83,11 +83,13 @@ class Examined(Record): parts: tuple[TracePart, ...] partial: bool cannot_assess: bool + error: str = "" class Investigation(Record): finding: FindingDraft | None parts: tuple[TracePart, ...] + error: str = "" ModelCall: TypeAlias = Callable[[ModelRequest], Awaitable[ModelResult]] @@ -98,6 +100,28 @@ ReportProgress: TypeAlias = Callable[[str, Coverage], Awaitable[None]] ResponseT = TypeVar("ResponseT", bound=Record) +class ValidationIssue(Record): + type: str + loc: tuple[str | int, ...] + msg: str + + +def validation_details(error: ValidationError) -> str: + issues: Final = TypeAdapter(tuple[ValidationIssue, ...]).validate_json( + error.json(include_input=False, include_context=False, include_url=False) + ) + return "\n".join( + f"{'.'.join(str(part) for part in issue.loc) or '$'}: {issue.msg} [{issue.type}]" + if issue.type != "extra_forbidden" + else "Unexpected field: Extra inputs are not permitted [extra_forbidden]" + for issue in issues + ) + + +class AnalysisResponseError(ValueError): + pass + + async def structured_response( request: ModelRequest, schema: type[ResponseT], @@ -107,6 +131,8 @@ async def structured_response( response: Final = await model(request) try: parsed: Final = schema.model_validate_json(response.content) + if response.finish_reason: + raise ValueError(f"Model did not finish its response (finish_reason={response.finish_reason})") invalid: Final = validate(parsed) if invalid: raise ValueError(invalid) @@ -124,11 +150,34 @@ async def structured_response( } ) ) - corrected: Final = schema.model_validate_json((await model(repair)).content) - remaining: Final = validate(corrected) - if remaining: - raise ValueError(remaining) - return corrected + repaired: Final = await model(repair) + try: + corrected: Final = schema.model_validate_json(repaired.content) + if repaired.finish_reason: + raise ValueError(f"Model did not finish its response (finish_reason={repaired.finish_reason})") + remaining: Final = validate(corrected) + if remaining: + raise ValueError(remaining) + return corrected + except ValueError as error: + stage: Final = MappingProxyType( + { + "extract": "Reading executions", + "cluster": "Grouping observations", + "investigate": "Checking original evidence", + } + )[request.purpose] + detail: Final = validation_details(error) if isinstance(error, ValidationError) else str(error) + stopped: Final = ( + " Model output was truncated (finish_reason=length)." + if repaired.finish_reason == "length" + else " Model output was blocked (finish_reason=content_filter)." + if repaired.finish_reason == "content_filter" + else "" + ) + raise AnalysisResponseError( + f"{stage} failed: {schema.__name__} response invalid after 2 attempts.{stopped}\n{detail}" + ) from error def evidence_valid(evidence: Evidence, parts: tuple[TracePart, ...]) -> bool: @@ -203,8 +252,15 @@ async def extract(claim: Claim, execution: Execution, read: ReadContent, model: with trace_store() as store: try: return await extract_stored(claim, execution, read, model, store) - except ValidationError: - return Examined(execution=execution, observations=(), parts=(), partial=True, cannot_assess=True) + except (ValidationError, AnalysisResponseError) as error: + return Examined( + execution=execution, + observations=(), + parts=(), + partial=True, + cannot_assess=True, + error=validation_details(error) if isinstance(error, ValidationError) else str(error), + ) async def extract_stored( @@ -254,7 +310,7 @@ async def extract_stored( for p in (first_root,) if p is not None ), - "read_evidence": tuple(p.model_dump() for p in additional[-2:]), + "read_evidence": tuple(p.model_dump() for p in additional), "previous_observations": tuple(o.model_dump() for o in previous.observations), "completed_read_count": len(reads), "last_completed_read": reads[-1].model_dump() if reads else None, @@ -294,7 +350,7 @@ async def extract_stored( previous = response continue fetched = tuple([parts async for parts in concurrent_results(requested, fetch)]) - if not any(p.content and p not in additional for p in chain.from_iterable(fetched)): + if not any(p.content for p in chain.from_iterable(fetched)): must_decide = True previous = response continue @@ -355,8 +411,12 @@ async def investigate( with trace_store() as store: try: return await investigate_stored(claim, candidate, examined, read, model, store) - except ValidationError: - return Investigation(finding=None, parts=()) + except (ValidationError, AnalysisResponseError) as error: + return Investigation( + finding=None, + parts=(), + error=validation_details(error) if isinstance(error, ValidationError) else str(error), + ) async def investigate_stored( @@ -371,6 +431,8 @@ async def investigate_stored( navigation: ExecutionContent | None = None # rebind-ok: last fetched page reads: tuple[Decision, ...] = () # rebind-ok: track completed tool requests to detect loops observation_page = 0 # rebind-ok: model controls navigation through observations + evidence_page = 0 # rebind-ok: navigate all content in the fetched evidence batch + evidence_seen = frozenset((0,)) # rebind-ok: reset navigation history when evidence changes catalog_page = 0 # rebind-ok: model controls navigation through the run catalog feedback_page = 0 # rebind-ok: navigate bounded prior finding pages feedback: Final = feedback_pages(claim, candidate.check_id) @@ -381,6 +443,7 @@ async def investigate_stored( navigation: ExecutionContent | None, reads: tuple[Decision, ...], observation_page: int, + evidence_page: int, catalog_page: int, feedback_page: int, stalled: bool, @@ -411,7 +474,7 @@ async def investigate_stored( ) ) bounded: Final = partition_content(prioritized, 30000) - evidence: Final = bounded[0] if bounded else () + evidence: Final = bounded[evidence_page] if evidence_page < len(bounded) else () catalog_batches: Final = partition_items( (*relevant, *(item for item in examined if item not in relevant)), lambda item: len(item.execution.model_dump_json()), @@ -452,13 +515,13 @@ async def investigate_stored( "feedback_page": feedback_page, "feedback_pages": len(feedback), "evidence": tuple(p.model_dump() for p in evidence), + "evidence_page": evidence_page, + "evidence_pages": len(bounded), "must_decide": stalled, "last_read": navigation.model_dump(exclude=MappingProxyType({"parts": True})) if navigation else None, }, ensure_ascii=False, ) - if len(prompt) > 100000: - return Investigation(finding=None, parts=evidence) request: Final = ModelRequest(purpose="investigate", prompt=prompt) decision: Final = await investigation_decision(request, model, 1 if stalled else 2) if decision.action == "submit" and decision.finding: @@ -479,11 +542,12 @@ async def investigate_stored( ) ): return Investigation(finding=finding, parts=evidence) - if stalled or decision.action not in ("read", "observations", "catalog", "feedback"): + if stalled or decision.action not in ("read", "evidence", "observations", "catalog", "feedback"): return Investigation(finding=None, parts=evidence) page_count: Final = MappingProxyType( { "observations": len(supporting_batches), + "evidence": len(bounded), "catalog": len(catalog_batches), "feedback": len(feedback), } @@ -497,13 +561,20 @@ async def investigate_stored( ) while True: step_result = await decide( - additional, navigation, reads, observation_page, catalog_page, feedback_page, stalled + additional, navigation, reads, observation_page, evidence_page, catalog_page, feedback_page, stalled ) if isinstance(step_result, Decision) and step_result.action == "inconclusive": stalled = True continue if isinstance(step_result, Investigation): return step_result + if step_result.action == "evidence": + if step_result.page in evidence_seen: + stalled = True + else: + evidence_page = step_result.page + evidence_seen = evidence_seen | frozenset((evidence_page,)) + continue if any( (r.action, r.execution_id, r.cursor, r.offset, r.page) == (step_result.action, step_result.execution_id, step_result.cursor, step_result.offset, step_result.page) @@ -514,16 +585,20 @@ async def investigate_stored( reads = (*reads, step_result) if step_result.action == "observations": observation_page = step_result.page + evidence_page = 0 + evidence_seen = frozenset((0,)) elif step_result.action == "catalog": catalog_page = step_result.page elif step_result.action == "feedback": feedback_page = step_result.page elif any(e.execution.id == step_result.execution_id for e in examined): navigation = await read(step_result.execution_id or "", step_result.cursor, step_result.offset) - if not any(p.content and p not in additional for p in navigation.parts): + if not any(p.content for p in navigation.parts): stalled = True store.add_reads(navigation.parts) additional = navigation.parts + evidence_page = 0 + evidence_seen = frozenset((0,)) else: return Investigation(finding=None, parts=additional) @@ -619,7 +694,11 @@ async def _analyze_sample( await progress("Grouping observations", coverage) observations: Final = tuple(chain.from_iterable(item.observations for item in examined)) if not observations: - return Result(coverage=coverage, assessments=assessments) + return Result( + coverage=coverage, + assessments=assessments, + error="\n\n".join(dict.fromkeys(item.error for item in examined if item.error)), + ) batches: Final = observation_batches(observations) grouping: Final = coverage.model_copy(update=MappingProxyType({"grouping_batches": len(batches)})) clusters: Final = await cluster_batches(batches, limited_model, progress, grouping) @@ -638,6 +717,7 @@ async def _analyze_sample( return Result( findings=tuple(item.finding for item in investigated if item.finding is not None), assessments=assessments, + error="\n\n".join(dict.fromkeys(item.error for item in (*examined, *investigated) if item.error)), coverage=investigating.model_copy( update=MappingProxyType( {"investigated": len(candidates), "inconclusive": sum(item.finding is None for item in investigated)} @@ -657,7 +737,7 @@ async def cluster_batches( Candidate( check_id=o.check_id, kind=o.kind, - title=o.summary[:160], + title=o.summary, hypothesis=f"{o.kind}: {o.summary}", execution_ids=tuple(sorted(frozenset(e.execution_id for e in o.evidence))), ) diff --git a/litellm/proxy/lens/billing.py b/litellm/proxy/lens/billing.py index 8c1c691b87f..9adc0c0fe7b 100644 --- a/litellm/proxy/lens/billing.py +++ b/litellm/proxy/lens/billing.py @@ -1,4 +1,5 @@ -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Callable, Mapping +from contextlib import AbstractAsyncContextManager from typing import Final import orjson @@ -32,7 +33,7 @@ async def validate_key(key_id: str | None) -> UserAPIKeyAuth | None: async def complete( - key_id: str, data: dict[str, object], reserve: Callable[[], Awaitable[None]], incoming: Request + key_id: str, data: dict[str, object], reserve: Callable[[], AbstractAsyncContextManager[None]], incoming: Request ) -> tuple[ModelResponse, float | None]: from litellm.proxy import proxy_server from litellm.proxy.proxy_server import llm_router, proxy_config, proxy_logging_obj, version @@ -67,23 +68,23 @@ async def complete( ) try: auth: Final = await authorize_internal_virtual_key(key_id, request, data) - await reserve() processor: Final = ProxyBaseLLMRequestProcessing(data=data) fastapi_response: Final = Response() try: - response: Final = TypeAdapter(ModelResponse).validate_python( - await processor.base_process_llm_request( - request=request, - fastapi_response=fastapi_response, - user_api_key_dict=auth, - route_type="acompletion", - proxy_logging_obj=proxy_logging_obj, - general_settings=TypeAdapter(dict[str, object]).validate_python(proxy_server.general_settings), # pyright: ignore[reportUnknownMemberType] # Validate the legacy untyped config at the request boundary - proxy_config=proxy_config, - llm_router=llm_router, - version=version, + async with reserve(): + response: Final = TypeAdapter(ModelResponse).validate_python( + await processor.base_process_llm_request( + request=request, + fastapi_response=fastapi_response, + user_api_key_dict=auth, + route_type="acompletion", + proxy_logging_obj=proxy_logging_obj, + general_settings=TypeAdapter(dict[str, object]).validate_python(proxy_server.general_settings), # pyright: ignore[reportUnknownMemberType] # Validate the legacy untyped config at the request boundary + proxy_config=proxy_config, + llm_router=llm_router, + version=version, + ) ) - ) billed: Final = fastapi_response.headers.get("x-litellm-response-cost") return response, float(billed) if billed not in (None, "", "None") else None except Exception as exc: diff --git a/litellm/proxy/lens/endpoints.py b/litellm/proxy/lens/endpoints.py index dc0ae8c985c..a20258853c0 100644 --- a/litellm/proxy/lens/endpoints.py +++ b/litellm/proxy/lens/endpoints.py @@ -7,11 +7,12 @@ from types import MappingProxyType from typing import Annotated, Final, TypeAlias from uuid import uuid4 -from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import AwareDatetime, BaseModel, Field -from litellm.proxy._types import LitellmUserRoles, ModelAccessDeniedProxyException, UserAPIKeyAuth +from litellm.litellm_core_utils.secret_redaction import redact_internal_details +from litellm.proxy._types import LitellmUserRoles, ModelAccessDeniedProxyException, ProxyException, UserAPIKeyAuth from litellm.proxy.auth.auth_checks import can_key_call_model from litellm.proxy.auth.resolvers.exceptions import KeyNotFoundError from litellm.proxy.auth.user_api_key_auth import user_api_key_auth @@ -28,6 +29,7 @@ from litellm.proxy.lens.models import ( Lens, LensList, LensSettings, + LookbackHours, ModelRequest, ModelResult, Progress, @@ -324,18 +326,23 @@ class Preview(BaseModel): as_of: AwareDatetime | None = None offset: int = Field(default=0, ge=0) settings: LensSettings - lookback_hours: int = Field(default=24, ge=1, le=8760) + lookback_hours: LookbackHours = 24 @router.post("/preview/sample", response_model=Sample) async def preview_sample(body: Preview, auth: Auth, storage: StorageDep) -> Sample: validate_selection(body.settings) now: Final = min(body.as_of or datetime.now(timezone.utc), datetime.now(timezone.utc)) + try: + start: Final = int((now - timedelta(hours=body.lookback_hours)).timestamp() * 1000) + end: Final = int((now - timedelta(minutes=2)).timestamp() * 1000) + except (OverflowError, ValueError) as error: + raise HTTPException(422, "Preview window exceeds the supported calendar range") from error return await source_reader(storage).sample( user_scope(auth), body.settings, - int((now - timedelta(hours=body.lookback_hours)).timestamp() * 1000), - int((now - timedelta(minutes=2)).timestamp() * 1000), + start, + end, offset=body.offset, preview=True, ) @@ -346,7 +353,7 @@ class WorkerBilling(BaseModel): class WorkerName(WorkerBilling): - name: str = Field(default="Lens worker", min_length=1, max_length=100) + name: str = Field(default="Lens worker", min_length=1) @router.post("/workers/register", response_model=WorkerCreated) @@ -395,7 +402,7 @@ async def revoke_worker(worker_id: str, auth: Auth) -> bool: @router.post("/worker/claim", response_model=Claim | None) async def claim(worker: WorkerAuth, protocol_version: int = 1) -> Claim | None: - if protocol_version != 2: + if protocol_version not in (2, 3): raise HTTPException(409, "Upgrade the Lens worker using the current Connect worker command") if worker.analysis_key_id is None: raise HTTPException(409, "Assign an analysis key to this worker in Lens setup") @@ -491,12 +498,31 @@ async def content( return await source_reader(storage).content(lens.scope, execution, cursor, offset) +def model_failure(error: HTTPException | ProxyException) -> HTTPException: + if isinstance(error, ProxyException): + status: Final = int(error.code) if error.code.isdigit() else 500 + return HTTPException(status, {"lens_error": redact_internal_details(error.message)}, headers=error.headers) + if isinstance(error.detail, str): + return HTTPException( + error.status_code, {"lens_error": redact_internal_details(error.detail)}, headers=error.headers + ) + return error + + @router.post("/worker/{lens_id}/{job_id}/model", response_model=ModelResult) -async def model(lens_id: str, job_id: str, body: ModelRequest, worker: WorkerAuth, request: Request) -> ModelResult: +async def model( + lens_id: str, job_id: str, body: ModelRequest, worker: WorkerAuth, request: Request, response: Response +) -> ModelResult: from litellm.proxy.lens.inference import analyze lens, job = await assigned(lens_id, job_id, worker) - return await analyze(repository(), lens, job, worker, body, request) + try: + completion: Final = await analyze(repository(), lens, job, worker, body, request) + except (ProxyException, HTTPException) as error: + raise model_failure(error) from error + if completion.finish_reason: + response.headers["x-litellm-lens-finish-reason"] = completion.finish_reason + return completion @router.post("/worker/{lens_id}/{job_id}/result", response_model=Lens) diff --git a/litellm/proxy/lens/inference.py b/litellm/proxy/lens/inference.py index b8a9d7754ae..7208504cd1f 100644 --- a/litellm/proxy/lens/inference.py +++ b/litellm/proxy/lens/inference.py @@ -1,3 +1,5 @@ +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from datetime import datetime, timezone from types import MappingProxyType from typing import Final @@ -9,6 +11,7 @@ import litellm from litellm.exceptions import ModelNotMappedError from litellm.integrations.clickhouse.context import lens_analysis from litellm.litellm_core_utils.initialize_dynamic_callback_params import inherit_message_logging_privacy +from litellm.litellm_core_utils.token_counter import get_modified_max_tokens from litellm.proxy.lens.billing import complete, validate_key from litellm.proxy.lens.models import Job, Lens, ModelRequest, ModelResult, Worker from litellm.proxy.lens.repository import LensRepository @@ -21,11 +24,19 @@ class DeploymentParams(BaseModel): model: str input_cost_per_token: float | None = None output_cost_per_token: float | None = None + max_tokens: int | None = Field(default=None, gt=0) + max_completion_tokens: int | None = Field(default=None, gt=0) + + +class ModelCapacity(BaseModel): + model_config = ConfigDict(extra="ignore") + max_output_tokens: int | None = Field(default=None, gt=0) class Deployment(BaseModel): model_config = ConfigDict(extra="ignore") litellm_params: DeploymentParams + model_info: ModelCapacity = ModelCapacity() class Message(BaseModel): @@ -36,6 +47,7 @@ class Message(BaseModel): class Choice(BaseModel): model_config = ConfigDict(extra="ignore") message: Message + finish_reason: str | None = None class Completion(BaseModel): @@ -88,6 +100,36 @@ def deployment_prices(deployment: Deployment) -> Prices: ) from exc +def catalog_capacity(model: str) -> ModelCapacity: + try: + return ModelCapacity.model_validate(litellm.get_model_info(model=model)) + except (ModelNotMappedError, ValueError): + return ModelCapacity() + + +def output_tokens(deployment: Deployment, prompt: str | None = None) -> int: + params: Final = deployment.litellm_params + configured: Final = params.max_completion_tokens or params.max_tokens or deployment.model_info.max_output_tokens + capacity: Final = configured or catalog_capacity(params.model).max_output_tokens + if capacity is None: + raise HTTPException( + 400, + f"Output capacity is unknown for {params.model}. Set model_info.max_output_tokens to the model's " + "supported output capacity or configure max_tokens on its deployment.", + ) + if prompt is None: + return capacity + adjusted: Final = get_modified_max_tokens( + model=params.model, + base_model=params.model, + messages=[{"role": "system", "content": _SYSTEM}, {"role": "user", "content": prompt}], + user_max_tokens=capacity, + buffer_perc=0, + buffer_num=0, + ) + return adjusted if adjusted is not None else capacity + + def quote(deployments: tuple[Deployment, ...], prompt: str) -> float: prices: Final = tuple(deployment_prices(d) for d in deployments) input_rate: Final = max( @@ -102,7 +144,15 @@ def quote(deployments: tuple[Deployment, ...], prompt: str) -> float: ) for p in prices ) - return ((len((prompt + _SYSTEM).encode()) + 1024) * input_rate + 4096 * output_rate) * 2 + output: Final = min(output_tokens(d, prompt) for d in deployments) + input_tokens: Final = max( + litellm.token_counter( + model=d.litellm_params.model, + messages=[{"role": "system", "content": _SYSTEM}, {"role": "user", "content": prompt}], + ) + for d in deployments + ) + return input_tokens * input_rate + output * output_rate async def analyze( @@ -142,37 +192,7 @@ async def analyze( current, active.model_copy(update=MappingProxyType({"cost": active.cost + estimate})) ).model_copy(update=MappingProxyType({"spent": current.spent + estimate})) - async def reserve_budget() -> None: - if await repo.update(lens.id, reserve) is None: - raise HTTPException(409, "Could not reserve analysis budget") - - data: Final[dict[str, object]] = { # mutable-ok: proxy processing enriches request data - "model": job.settings.model, - "messages": [ - {"role": "system", "content": _SYSTEM}, - {"role": "user", "content": body.prompt}, - ], - "max_tokens": 4096, - "stream": False, - "timeout": 120, - "num_retries": 0, - "disable_fallbacks": True, - "response_format": {"type": "json_object"}, - "metadata": { - "tags": ["litellm-lens"], - "lens_id": lens.id, - "lens_run_id": job.id, - "lens_worker_id": worker.id, - "user_api_key_team_id": team_id, - }, - } - - with lens_analysis(), inherit_message_logging_privacy(True): - response, billed_cost = await complete(worker.analysis_key_id, data, reserve_budget, request) - parsed: Final = Completion.model_validate_json(response.model_dump_json()) - cost: Final = billed_cost if billed_cost is not None else completion_charge(deployments, response, estimate) - - def settle(e: Lens) -> Lens: + def settle(e: Lens, cost: float) -> Lens: charged: Final = next((j for j in e.jobs if j.id == job.id), None) adjusted: Final = ( e.model_copy(update=MappingProxyType({"spent": max(0, e.spent - estimate + cost)})) @@ -187,8 +207,50 @@ async def analyze( else adjusted ) - await repo.update(lens.id, settle) - return ModelResult(content=parsed.choices[0].message.content or "{}", cost=cost) + @asynccontextmanager + async def reserve_budget() -> AsyncIterator[None]: + if await repo.update(lens.id, reserve) is None: + raise HTTPException(409, "Could not reserve analysis budget") + try: + yield + except BaseException: + await repo.update(lens.id, lambda e: settle(e, 0)) + raise + + data: Final[dict[str, object]] = { # mutable-ok: proxy processing enriches request data + "model": job.settings.model, + "messages": [ + {"role": "system", "content": _SYSTEM}, + {"role": "user", "content": body.prompt}, + ], + "max_tokens": min(output_tokens(d, body.prompt) for d in deployments), + "stream": False, + "num_retries": 0, + "disable_fallbacks": True, + "response_format": {"type": "json_object"}, + "metadata": { + "tags": ["litellm-lens"], + "lens_id": lens.id, + "lens_run_id": job.id, + "lens_worker_id": worker.id, + "user_api_key_team_id": team_id, + }, + } + + with lens_analysis(), inherit_message_logging_privacy(True): + response, billed_cost = await complete(worker.analysis_key_id, data, reserve_budget, request) + cost: Final = billed_cost if billed_cost is not None else completion_charge(deployments, response, estimate) + + await repo.update(lens.id, lambda e: settle(e, cost)) + parsed: Final = Completion.model_validate_json(response.model_dump_json()) + choice: Final = parsed.choices[0] + return ModelResult( + content=choice.message.content or "", + cost=cost, + finish_reason="length" + if choice.finish_reason == "length" + else ("content_filter" if choice.finish_reason == "content_filter" else None), + ) def completion_charge(deployments: tuple[Deployment, ...], response: ModelResponse, estimate: float) -> float: diff --git a/litellm/proxy/lens/models.py b/litellm/proxy/lens/models.py index 7add39e41be..d967d18d0d1 100644 --- a/litellm/proxy/lens/models.py +++ b/litellm/proxy/lens/models.py @@ -1,7 +1,27 @@ -from datetime import datetime -from typing import Final, Literal +from datetime import datetime, timedelta, timezone +from typing import Annotated, Final, Literal, TypeAlias -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import AfterValidator, BaseModel, ConfigDict, Field, model_validator + + +def calendar_lookback(hours: int) -> int: + try: + datetime.now(timezone.utc) - timedelta(hours=hours) + except OverflowError as error: + raise ValueError("Lookback exceeds the supported calendar range") from error + return hours + + +def calendar_interval(minutes: int) -> int: + try: + datetime.now(timezone.utc) + timedelta(minutes=minutes) + except OverflowError as error: + raise ValueError("Interval exceeds the supported calendar range") from error + return minutes + + +LookbackHours: TypeAlias = Annotated[int, Field(ge=1), AfterValidator(calendar_lookback)] +IntervalMinutes: TypeAlias = Annotated[int, Field(ge=1), AfterValidator(calendar_interval)] class Record(BaseModel): @@ -15,34 +35,34 @@ class Scope(Record): class MetadataFilter(Record): - key: str = Field(min_length=1, max_length=200) - value: str = Field(min_length=1, max_length=500) + key: str = Field(min_length=1) + value: str = Field(min_length=1) class Check(Record): - id: str = Field(min_length=1, max_length=80) - instruction: str = Field(min_length=3, max_length=3000) + id: str = Field(min_length=1) + instruction: str = Field(min_length=3) enabled: bool = True class LensSettings(Record): - name: str = Field(min_length=1, max_length=100) - context: str = Field(default="", max_length=6000) + name: str = Field(min_length=1) + context: str = Field(default="") source: Literal["traces", "requests", "both"] = "traces" - lookback_hours: int = Field(default=24, ge=1, le=8760) - service: str = Field(default="", max_length=200) - agent_name: str = Field(default="", max_length=200) - filters: tuple[MetadataFilter, ...] = Field(default=(), max_length=8) + lookback_hours: LookbackHours = 24 + service: str = Field(default="") + agent_name: str = Field(default="") + filters: tuple[MetadataFilter, ...] = Field(default=()) checks: tuple[Check, ...] = () - model: str = Field(min_length=1, max_length=200) + model: str = Field(min_length=1) enabled: bool = True - interval_minutes: int = Field(default=15, ge=1, le=10080) + interval_minutes: IntervalMinutes = 15 sample_size: int | None = Field(default=None, ge=1) sample_percent: float = Field(default=100, gt=0, le=100, allow_inf_nan=False) concurrency: int = Field(default=8, ge=1) team_id: str = "" execution_ids: tuple[str, ...] = () - monthly_budget: float = Field(default=100, gt=0, le=100000, allow_inf_nan=False) + monthly_budget: float = Field(default=100, gt=0, allow_inf_nan=False) @model_validator(mode="after") def unique_checks(self) -> "LensSettings": @@ -72,32 +92,32 @@ class LensSettings(Record): class Evidence(Record): execution_id: str span_id: str - quote: str = Field(min_length=1, max_length=1000) + quote: str = Field(min_length=1) role: Literal["support", "counterexample"] = "support" class AgentTestCase(Record): - input: str = Field(min_length=1, max_length=1000) - expected: str = Field(min_length=1, max_length=1000) + input: str = Field(min_length=1) + expected: str = Field(min_length=1) class IssueBrief(Record): - problem: str = Field(min_length=10, max_length=400) - user_goal: str = Field(min_length=3, max_length=400) - what_happened: str = Field(min_length=3, max_length=1500) - test_cases: tuple[AgentTestCase, ...] = Field(min_length=1, max_length=5) + problem: str = Field(min_length=10) + user_goal: str = Field(min_length=3) + what_happened: str = Field(min_length=3) + test_cases: tuple[AgentTestCase, ...] = Field(min_length=1) class FindingDraft(Record): - title: str = Field(min_length=3, max_length=160) - description: str = Field(min_length=10, max_length=4000) + title: str = Field(min_length=3) + description: str = Field(min_length=10) check_id: str kind: Literal["issue", "pattern"] = "issue" priority: Literal["high", "medium", "low"] = "medium" - suggestion: str = Field(default="", max_length=2000) - limitation: str = Field(default="", max_length=600) + suggestion: str = Field(default="") + limitation: str = Field(default="") brief: IssueBrief | None = None - evidence: tuple[Evidence, ...] = Field(min_length=1, max_length=20) + evidence: tuple[Evidence, ...] = Field(min_length=1) existing_finding_id: str | None = None @@ -228,12 +248,12 @@ class LensList(Record): class RunRequest(Record): settings: LensSettings | None = None - lookback_hours: int | None = Field(default=None, ge=1, le=8760) + lookback_hours: LookbackHours | None = None class FindingUpdate(Record): status: Literal["open", "resolved", "dismissed"] - reason: str = Field(default="", max_length=2000) + reason: str = Field(default="") class Claim(Record): @@ -243,7 +263,7 @@ class Claim(Record): class Progress(Record): - stage: str = Field(max_length=100) + stage: str = Field() coverage: Coverage = Coverage() @@ -251,14 +271,15 @@ class Result(Record): assessments: tuple[RunAssessment, ...] = () findings: tuple[FindingDraft, ...] = () coverage: Coverage - error: str = Field(default="", max_length=1000) + error: str = Field(default="") class ModelRequest(Record): - prompt: str = Field(min_length=1, max_length=100000) + prompt: str = Field(min_length=1) purpose: Literal["extract", "cluster", "investigate"] class ModelResult(Record): content: str cost: float + finish_reason: Literal["length", "content_filter"] | None = Field(default=None, exclude=True) diff --git a/litellm/proxy/lens/prompts/investigate.md b/litellm/proxy/lens/prompts/investigate.md index edf795de462..ae3dcfba6ec 100644 --- a/litellm/proxy/lens/prompts/investigate.md +++ b/litellm/proxy/lens/prompts/investigate.md @@ -10,6 +10,8 @@ Return action='read' with execution_id, cursor (span ID; default empty), offset Reads return up to 40 spans; advance cursor from next_cursor for more spans or offset by 8000 for longer content; offset=1 reads original beginning after an abbreviated excerpt. Read any execution in the supplied catalog. Use action='catalog' or 'observations' with page to fetch another page of runs or supporting observations. +Use action='evidence' with page to read the remaining content in a fetched batch; evidence_pages includes every supplied span. +Read needed evidence pages before advancing the span cursor. Evidence pages reset to zero after a read or observation-page change. Use action=feedback to read prior findings and dismissal reasons only when feedback_pages>1. The current page is already supplied; feedback_pages=0 means no prior findings or feedback exist, so do not request feedback. Request only page numbers below the corresponding page count. @@ -19,16 +21,16 @@ Mark quotes from runs that demonstrate the opposite behavior as counterexample, Include at least one supporting quote. Never put internal run aliases in prose; the evidence links identify the runs. Write for a busy person, in plain English. -Title: a short, concrete outcome in at most 12 words. -Description: one or two short sentences saying what happened and why it matters, at most 60 words. -Put uncertainty or counterexamples in limitation, not in the main description; use at most 40 words. -Suggestion: one specific action, at most 25 words, or empty if no action is needed. +Title: a short, concrete outcome. +Description: one or two short sentences saying what happened and why it matters. +Put uncertainty or counterexamples in limitation, not in the main description. +Suggestion: one specific action, or empty if no action is needed. For issues, also return brief, which describes the failure so anyone can reproduce and verify it without access to the agent's code. Scope what went wrong from the evidence: compare each failed or empty tool result with the tools, permissions, working directory, and configuration visible in the recorded requests, and name the most specific cause the evidence supports. brief.problem: the root cause in one or two sentences. brief.user_goal: what the end user was trying to achieve. brief.what_happened: what the agent actually output or did, quoting the recorded output where possible. -brief.test_cases: one to five user inputs drawn from the evidence, each with the behavior a correct agent should show. +brief.test_cases: user inputs drawn from the evidence, each with the behavior a correct agent should show. Do not prescribe code or configuration changes in brief. Omit brief for patterns. Avoid jargon such as document-borne, visible noncompliance, instruction-bearing, or evaluator-directed. diff --git a/litellm/proxy/lens/prompts/review.md b/litellm/proxy/lens/prompts/review.md index 727c9ad55ed..d1a5e590dfd 100644 --- a/litellm/proxy/lens/prompts/review.md +++ b/litellm/proxy/lens/prompts/review.md @@ -20,7 +20,6 @@ Respect prior feedback about accepted behavior, but do not suppress different pr Request reads with span_id and offset=0 for initial evidence. If an excerpt omits content, offset=1 reads the original beginning; later offsets advance by 8000 characters through the original stored span. Do not repeat a completed read. -At most two reads per turn. Return observations using an enabled check ID, exact quotes, and the correct execution_id/span_id. Never quote an omission marker or join text from either side of one. If you need more evidence, return reads; otherwise return reads=[] and your final observations. diff --git a/litellm/proxy/lens/sources.py b/litellm/proxy/lens/sources.py index 9dbe635e348..9af0f3679b6 100644 --- a/litellm/proxy/lens/sources.py +++ b/litellm/proxy/lens/sources.py @@ -126,7 +126,7 @@ class SourceReader: metadata=tuple( MetadataFilter(key=k, value=v) for k, v in row.attributes - if k != "litellm.api_key_hash" and 0 < len(k) <= 200 and 0 < len(v) <= 500 + if k != "litellm.api_key_hash" and k and v ), ) for row in rows diff --git a/litellm/proxy/lens/worker.py b/litellm/proxy/lens/worker.py index 051b4a09392..f9e746489d9 100644 --- a/litellm/proxy/lens/worker.py +++ b/litellm/proxy/lens/worker.py @@ -3,14 +3,13 @@ import logging import os import sqlite3 from collections.abc import Awaitable, Callable -from contextlib import suppress from types import MappingProxyType from typing import Final import httpx from pydantic import BaseModel, ConfigDict, ValidationError -from .analysis import analyze_sample +from .analysis import AnalysisResponseError, analyze_sample, validation_details from .models import Claim, Coverage, ExecutionContent, ModelRequest, ModelResult, Progress, Result, Sample logger: Final = logging.getLogger("litellm.lens.worker") @@ -27,7 +26,21 @@ class ClaimIdentity(BaseModel): job: ClaimedJobIdentity +class PublicModelError(BaseModel): + model_config = ConfigDict(extra="ignore") + lens_error: str + + +class ModelErrorEnvelope(BaseModel): + model_config = ConfigDict(extra="ignore") + detail: PublicModelError + + def failure_message(error: Exception) -> str: + if isinstance(error, AnalysisResponseError): + return str(error) + if isinstance(error, ValidationError): + return f"Invalid {error.title} response (ValidationError):\n{validation_details(error)}" if isinstance(error, (OSError, sqlite3.Error)): return "Worker temporary storage failed. Increase its capacity or reduce analysis parallelism." if isinstance(error, httpx.TimeoutException): @@ -46,6 +59,12 @@ def failure_message(error: Exception) -> str: else "Worker request" ) status: Final = error.response.status_code + if path.endswith("/model"): + try: + diagnostic: Final = ModelErrorEnvelope.model_validate_json(error.response.content) + return f"Model request failed (HTTP {status}):\n{diagnostic.detail.lens_error}" + except ValueError: + pass guidance: Final = MappingProxyType( { 400: "Check the configured model and whether the worker's billing key is enabled.", @@ -62,15 +81,33 @@ def failure_message(error: Exception) -> str: class LensWorker: - def __init__(self, client: httpx.AsyncClient, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep) -> None: + def __init__( + self, + client: httpx.AsyncClient, + sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, + heartbeat_wait: Callable[[float], Awaitable[None]] = asyncio.sleep, + ) -> None: self.client: Final = client self.sleep: Final = sleep + self.heartbeat_wait: Final = heartbeat_wait async def model_request(self, path: str, body: ModelRequest, attempt: int = 0) -> ModelResult: try: - result: Final = await self.client.post(path, json=body.model_dump()) + timeout: Final = httpx.Timeout( + None, + connect=self.client.timeout.connect, + write=self.client.timeout.write, + pool=self.client.timeout.pool, + ) + result: Final = await self.client.post(path, json=body.model_dump(), timeout=timeout) result.raise_for_status() - return ModelResult.model_validate(result.json()) + parsed: Final = ModelResult.model_validate(result.json()) + reason: Final = result.headers.get("x-litellm-lens-finish-reason") + return ( + parsed.model_copy(update=MappingProxyType({"finish_reason": reason})) + if reason in ("length", "content_filter") + else parsed + ) except (httpx.TransportError, httpx.HTTPStatusError) as exc: retryable: Final = not isinstance(exc, httpx.HTTPStatusError) or exc.response.status_code in ( 429, @@ -84,7 +121,7 @@ class LensWorker: return await self.model_request(path, body, attempt + 1) async def run_once(self) -> bool: - response: Final = await self.client.post("/lens/worker/claim", params=MappingProxyType({"protocol_version": 2})) + response: Final = await self.client.post("/lens/worker/claim", params=MappingProxyType({"protocol_version": 3})) response.raise_for_status() payload: Final = response.json() if payload is None: @@ -131,17 +168,30 @@ class LensWorker: async def heartbeat() -> None: while True: - await asyncio.sleep(30) - (await self.client.post(prefix + "/heartbeat")).raise_for_status() + await self.heartbeat_wait(30) + try: + (await self.client.post(prefix + "/heartbeat")).raise_for_status() + except (httpx.TransportError, httpx.HTTPStatusError) as exc: + if isinstance(exc, httpx.HTTPStatusError) and ( + exc.response.status_code < 500 and exc.response.status_code != 429 + ): + raise + logger.warning("Analysis %s heartbeat will retry (%s)", claim.job.id, type(exc).__name__) - pulse_task: Final = asyncio.create_task(heartbeat()) - try: + async def investigate() -> None: data: Final = await self.client.get(prefix + "/sample") data.raise_for_status() sample: Final = Sample.model_validate(data.json()) result: Final = await analyze_sample(claim, sample, read, model, progress) saved: Final = await self.client.post(prefix + "/result", json=result.model_dump(mode="json")) saved.raise_for_status() + + pulse_task: Final = asyncio.create_task(heartbeat()) + work_task: Final = asyncio.create_task(investigate()) + try: + finished, _ = await asyncio.wait((pulse_task, work_task), return_when=asyncio.FIRST_COMPLETED) + for task in finished: + await task except (httpx.HTTPError, ValueError, OSError, sqlite3.Error) as exc: message: Final = failure_message(exc) logger.warning("Analysis %s interrupted (%s)", claim.job.id, type(exc).__name__) @@ -152,8 +202,8 @@ class LensWorker: failed.raise_for_status() finally: pulse_task.cancel() - with suppress(asyncio.CancelledError, httpx.HTTPError): - await pulse_task + work_task.cancel() + await asyncio.gather(pulse_task, work_task, return_exceptions=True) return True diff --git a/tests/proxy_behavior/lens/test_lifecycle.py b/tests/proxy_behavior/lens/test_lifecycle.py index 77e1421675a..1dade952dd8 100644 --- a/tests/proxy_behavior/lens/test_lifecycle.py +++ b/tests/proxy_behavior/lens/test_lifecycle.py @@ -8,7 +8,7 @@ from uuid import uuid4 import pytest import pytest_asyncio -from fastapi import HTTPException, Request +from fastapi import HTTPException, Request, Response from fastapi.security import HTTPAuthorizationCredentials from pydantic import TypeAdapter @@ -57,6 +57,18 @@ async def lens_database() -> AsyncIterator[PrismaClient]: "model": "openai/lens-test-analysis", "api_key": "test-only", "mock_response": '{"observations":[]}', + "max_tokens": 16384, + "input_cost_per_token": 0.000001, + "output_cost_per_token": 0.000002, + }, + }, + { + "model_name": "lens-failing-analysis", + "litellm_params": { + "model": "openai/lens-failing-analysis", + "api_key": "test-only", + "mock_response": "litellm.RateLimitError", + "max_tokens": 16384, "input_cost_per_token": 0.000001, "output_cost_per_token": 0.000002, }, @@ -273,6 +285,7 @@ async def test_scan_lifecycle_persists_results_and_revokes_worker(lens_database: "client": ("127.0.0.1", 1234), } ), + response=Response(), ) assert '"observations"' in response.content with pytest.raises(HTTPException) as denied_ip: @@ -290,6 +303,7 @@ async def test_scan_lifecycle_persists_results_and_revokes_worker(lens_database: "client": ("192.0.2.1", 1234), } ), + response=Response(), ) assert denied_ip.value.status_code == 403 forwarded: Final = await endpoints.model( @@ -306,6 +320,7 @@ async def test_scan_lifecycle_persists_results_and_revokes_worker(lens_database: "client": ("192.0.2.100", 1234), } ), + response=Response(), ) assert '"observations"' in forwarded.content with pytest.raises(HTTPException) as spoofed_chain: @@ -323,6 +338,7 @@ async def test_scan_lifecycle_persists_results_and_revokes_worker(lens_database: "client": ("192.0.2.100", 1234), } ), + response=Response(), ) assert spoofed_chain.value.status_code == 403 charged: Final = await endpoints.get_lens(lens.id, worker.scope) @@ -387,3 +403,46 @@ async def test_scan_lifecycle_persists_results_and_revokes_worker(lens_database: await lens_database.db.execute_raw('DELETE FROM "LiteLLM_Lens" WHERE id=$1', lens.id) await lens_database.db.execute_raw('DELETE FROM "LiteLLM_LensWorker" WHERE id=$1', worker.id) await lens_database.db.execute_raw('DELETE FROM "LiteLLM_VerificationToken" WHERE token=$1', key_id) + + +@pytest.mark.asyncio +async def test_failed_model_requests_release_lens_budget_reservations(lens_database: PrismaClient) -> None: + admin: Final = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) + settings: Final = LensSettings( + name="Failed billing regression", model="lens-failing-analysis", context="Verify outcomes", enabled=False + ) + lens: Final = await endpoints.create_lens(settings, admin) + key_id: Final = hashlib.sha256(uuid4().bytes).hexdigest() + await lens_database.db.litellm_verificationtoken.create(data={"token": key_id, "models": [settings.model]}) + registration: Final = await endpoints.register_worker(endpoints.WorkerName(analysis_key_id=key_id), admin) + worker: Final = registration.worker + try: + claimed: Final = await endpoints.claim_candidate(lens, worker, datetime.now(timezone.utc)) + assert claimed is not None + for _ in range(3): + with pytest.raises(HTTPException) as failed: + await endpoints.model( + lens.id, + claimed.job.id, + ModelRequest(prompt="Return JSON", purpose="extract"), + worker, + Request( + { + "type": "http", + "scheme": "http", + "path": "/lens/worker/model", + "headers": [], + "client": ("127.0.0.1", 1234), + } + ), + response=Response(), + ) + assert failed.value.status_code == 429 + stored: Final = await endpoints.get_lens(lens.id, worker.scope) + assert stored.spent == 0 + assert stored.jobs[0].cost == 0 + finally: + await lens_database.db.execute_raw('DELETE FROM "LiteLLM_LensRun" WHERE lens_id=$1', lens.id) + await lens_database.db.execute_raw('DELETE FROM "LiteLLM_Lens" WHERE id=$1', lens.id) + await lens_database.db.execute_raw('DELETE FROM "LiteLLM_LensWorker" WHERE id=$1', worker.id) + await lens_database.db.execute_raw('DELETE FROM "LiteLLM_VerificationToken" WHERE token=$1', key_id) diff --git a/tests/unit/proxy/lens/test_analysis.py b/tests/unit/proxy/lens/test_analysis.py index d710bcce937..5231000e14b 100644 --- a/tests/unit/proxy/lens/test_analysis.py +++ b/tests/unit/proxy/lens/test_analysis.py @@ -398,23 +398,20 @@ async def test_investigator_keeps_final_outcome_ahead_of_repeated_model_history( "quote, check_id, accepted", [("timeout", "retries", True), ("invented quote", "retries", False), ("timeout", "unknown", False)], ) -async def test_oversized_model_evidence_is_retried_and_quotes_still_verified( +async def test_many_model_citations_are_accepted_but_quotes_are_still_verified( quote: str, check_id: str, accepted: bool ) -> None: execution: Final = Execution( id="run1", source="traces", trace_id="t", team_id="alpha", name="review", start_time="", span_count=1 ) part: Final = TracePart(execution_id="run1", span_id="span", name="tool", kind="tool", content="timeout") - attempts: Final = iter((8, 1)) + attempts: Final = iter((8,)) async def read(_execution_id: str, _cursor: str, _offset: int) -> ExecutionContent: return ExecutionContent(execution=execution, parts=(part,)) async def model(request: ModelRequest) -> ModelResult: count: Final = next(attempts) - if count == 1: - assert "validation errors" in request.prompt - assert '"max_length":6' in request.prompt evidence: Final = Evidence(execution_id="run1", span_id="span", quote=quote).model_dump_json() return ModelResult( content='{"observations":[{"check_id":"' @@ -434,9 +431,7 @@ async def test_oversized_model_evidence_is_retried_and_quotes_still_verified( @pytest.mark.asyncio async def test_invalid_model_output_has_only_one_repair_attempt() -> None: - from pydantic import ValidationError - - from litellm.proxy.lens.analysis import Extraction, structured_response + from litellm.proxy.lens.analysis import AnalysisResponseError, Extraction, structured_response attempts: Final = iter((1, 2)) @@ -444,7 +439,9 @@ async def test_invalid_model_output_has_only_one_repair_attempt() -> None: assert next(attempts, None) is not None, "Model repair exceeded its retry limit" return ModelResult(content="not JSON", cost=0) - with pytest.raises(ValidationError): + with pytest.raises( + AnalysisResponseError, match="Reading executions failed: Extraction response invalid after 2 attempts" + ): await structured_response(ModelRequest(purpose="extract", prompt="Extract observations"), Extraction, model) assert next(attempts, None) is None @@ -505,16 +502,18 @@ async def test_investigator_can_cite_a_later_page_or_offset(later_span: str) -> draft: Final = finding("run1").model_copy( update={"evidence": (Evidence(execution_id="run1", span_id=later_span, quote="timeout"),)} ) - decisions: Final = iter(("read", "submit")) + offsets: Final = iter((8000, 16000, None)) async def model(request: ModelRequest) -> ModelResult: - if next(decisions) == "read": - return ModelResult(content='{"action":"read","execution_id":"run1","offset":8000}', cost=0) + offset: Final = next(offsets) + if offset is not None: + return ModelResult(content=json.dumps({"action": "read", "execution_id": "run1", "offset": offset}), cost=0) + assert json.loads(request.prompt)["must_decide"] is False assert '"content": "timeout"' in request.prompt return ModelResult(content='{"action":"submit","finding":' + draft.model_dump_json() + "}", cost=0) async def read(execution_id: str, _cursor: str, offset: int) -> ExecutionContent: - assert execution_id == "run1" and offset == 8000 + assert execution_id == "run1" and offset in (8000, 16000) return ExecutionContent(execution=execution, parts=(later,)) claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) @@ -963,6 +962,7 @@ async def test_invalid_candidate_response_preserves_other_findings_and_reports_i ) assert tuple(result.finding for result in results if result.finding is not None) == (finding("run"),) assert sum(result.finding is None for result in results) == 1 + assert "[json_invalid]" in next(result.error for result in results if result.finding is None) assert max(counts.get_nowait() for _ in range(counts.qsize())) == 1 @@ -996,3 +996,203 @@ async def test_investigator_keeps_the_issue_brief() -> None: ) assert result.finding is not None assert result.finding.brief == draft.brief + + +@pytest.mark.asyncio +@pytest.mark.parametrize("finish_reason", (None, "length", "content_filter")) +async def test_grouping_failure_keeps_validation_details_without_model_content(finish_reason: str | None) -> None: + from litellm.proxy.lens.analysis import AnalysisResponseError, Clusters, structured_response + + async def model(_request: ModelRequest) -> ModelResult: + return ModelResult.model_validate( + {"content": '{"candidates":[{"title":"private trace"}]}', "cost": 0, "finish_reason": finish_reason} + ) + + with pytest.raises(AnalysisResponseError) as caught: + await structured_response(ModelRequest(purpose="cluster", prompt="private evidence"), Clusters, model) + message: Final = str(caught.value) + assert message.startswith("Grouping observations failed: Clusters response invalid after 2 attempts.") + assert "candidates.0.check_id: Field required [missing]" in message + assert "private" not in message + if finish_reason: + assert f"finish_reason={finish_reason}" in message + else: + assert "truncated" not in message + + +@pytest.mark.asyncio +async def test_truncated_but_valid_json_is_repaired_before_accepting_findings() -> None: + from litellm.proxy.lens.analysis import Clusters, structured_response + + outputs: Final = iter( + ( + ModelResult(content='{"candidates":[]}', cost=0, finish_reason="length"), + ModelResult(content='{"candidates":[]}', cost=0), + ) + ) + + async def model(_request: ModelRequest) -> ModelResult: + return next(outputs) + + assert await structured_response(ModelRequest(purpose="cluster", prompt="group"), Clusters, model) == Clusters() + assert next(outputs, None) is None + + +@pytest.mark.asyncio +async def test_large_context_and_long_verified_quotes_do_not_silently_end_investigation() -> None: + from litellm.proxy.lens.models import FindingDraft, LensSettings + + context: Final = "Read all recorded evidence. " * 5000 + long_quote: Final = "timeout detail " * 200 + execution: Final = Execution( + id="run", source="traces", trace_id="t", team_id="", name="task", start_time="", span_count=1 + ) + part: Final = TracePart(execution_id="run", span_id="span", name="tool", kind="tool", content=long_quote) + reviewed: Final = Examined(execution=execution, observations=(), parts=(part,), partial=False, cannot_assess=False) + expected: Final = FindingDraft.model_validate( + { + **finding("run").model_dump(), + "description": "Recorded failure detail. " * 300, + "evidence": [{"execution_id": "run", "span_id": "span", "quote": long_quote}], + } + ) + settings: Final = LensSettings.model_validate({**lens().settings.model_dump(), "context": context}) + claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job", settings=settings).jobs[0], findings=()) + + async def model(request: ModelRequest) -> ModelResult: + assert json.loads(request.prompt)["context"] == context + return ModelResult(content=json.dumps({"action": "submit", "finding": expected.model_dump()}), cost=0) + + async def read(_execution_id: str, _cursor: str, _offset: int) -> ExecutionContent: + pytest.fail("Already supplied evidence should not require a read") + + result: Final = await investigate( + claim, + Candidate(check_id="retries", title="Failure", hypothesis="Retry failed", execution_ids=("run",)), + (reviewed,), + read, + model, + ) + assert result.finding == expected + + +@pytest.mark.asyncio +async def test_reviewer_can_read_every_offset_of_a_long_span_before_deciding() -> None: + execution: Final = Execution( + id="run", source="traces", trace_id="t", team_id="", name="task", start_time="", span_count=1 + ) + original: Final = "trace evidence! " * 16000 + "late verified failure" + offsets: Final = SimpleQueue[int]() + seen: Final = SimpleQueue[str]() + + async def read(_execution_id: str, _cursor: str, offset: int) -> ExecutionContent: + offsets.put(offset) + content: Final = ( + "Preview; read for complete content" if offset == 0 else original[offset - 1 : offset - 1 + 8000] + ) + return ExecutionContent( + execution=execution, + parts=( + TracePart( + execution_id="run", + span_id="span", + name="agent", + kind="agent", + content=content, + truncated=offset == 0 or offset - 1 + 8000 < len(original), + ), + ), + ) + + async def model(request: ModelRequest) -> ModelResult: + payload: Final = json.loads(request.prompt) + read_count: Final = payload["completed_read_count"] + if read_count: + seen.put(payload["read_evidence"][0]["content"]) + if read_count * 8000 < len(original): + return ModelResult( + content=json.dumps({"reads": [{"span_id": "span", "offset": 1 + read_count * 8000}]}), cost=0 + ) + return ModelResult( + content=json.dumps( + { + "observations": [ + { + "check_id": "retries", + "summary": "Late failure", + "evidence": [{"execution_id": "run", "span_id": "span", "quote": "late verified failure"}], + } + ] + } + ), + cost=0, + ) + + claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) + result: Final = await extract(claim, execution, read, model) + assert "".join(seen.get_nowait() for _ in range(seen.qsize())) == original + assert tuple(offsets.get_nowait() for _ in range(offsets.qsize())) == (0, *range(1, len(original) + 1, 8000)) + assert result.observations[0].evidence[0].quote == "late verified failure" + assert not result.cannot_assess + + +@pytest.mark.asyncio +async def test_investigator_can_read_all_evidence_pages_across_successive_span_batches() -> None: + execution: Final = Execution( + id="run", source="traces", trace_id="t", team_id="", name="task", start_time="", span_count=80 + ) + parts: Final = tuple( + TracePart( + execution_id="run", + span_id=f"span{i:03}", + parent_span_id="root", + name=f"Step {i}", + kind="tool", + content="recorded evidence " * 400 + ("timeout" if i == 79 else "complete"), + ) + for i in range(80) + ) + seen: Final = SimpleQueue[str]() + read_cursors: Final = SimpleQueue[str]() + expected: Final = finding("run").model_copy( + update={"evidence": (Evidence(execution_id="run", span_id="span079", quote="timeout"),)} + ) + + async def read(_identity: str, cursor: str, _offset: int) -> ExecutionContent: + read_cursors.put(cursor) + assert cursor in ("", "span039") + return ExecutionContent( + execution=execution, + parts=parts[:40] if not cursor else parts[40:], + next_cursor="span039" if not cursor else None, + ) + + async def model(request: ModelRequest) -> ModelResult: + payload: Final = json.loads(request.prompt) + if not payload["completed_read_count"]: + return ModelResult(content=json.dumps({"action": "read", "execution_id": "run"}), cost=0) + for part in payload["evidence"]: + seen.put(part["span_id"]) + if payload["evidence_page"] + 1 < payload["evidence_pages"]: + return ModelResult(content=json.dumps({"action": "evidence", "page": payload["evidence_page"] + 1}), cost=0) + if payload["last_read"]["next_cursor"]: + return ModelResult( + content=json.dumps( + {"action": "read", "execution_id": "run", "cursor": payload["last_read"]["next_cursor"]} + ), + cost=0, + ) + return ModelResult(content=json.dumps({"action": "submit", "finding": expected.model_dump()}), cost=0) + + claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) + result: Final = await investigate( + claim, + Candidate(check_id="retries", title="Failure", hypothesis="Failure", execution_ids=("run",)), + (Examined(execution=execution, observations=(), parts=(), partial=False, cannot_assess=False),), + read, + model, + ) + assert result.finding == expected + assert result.error == "" + assert tuple(seen.get_nowait() for _ in range(seen.qsize())) == tuple(p.span_id for p in parts) + assert tuple(read_cursors.get_nowait() for _ in range(read_cursors.qsize())) == ("", "span039") diff --git a/tests/unit/proxy/lens/test_endpoints.py b/tests/unit/proxy/lens/test_endpoints.py index a1441b34ffa..40a6fd55044 100644 --- a/tests/unit/proxy/lens/test_endpoints.py +++ b/tests/unit/proxy/lens/test_endpoints.py @@ -165,3 +165,45 @@ def test_regular_keys_cannot_read_lens_results(role: LitellmUserRoles | None) -> with pytest.raises(HTTPException) as error: user_scope(auth) assert error.value.status_code == 403 + + +@pytest.mark.parametrize("provider", (False, True)) +def test_model_errors_reach_worker_with_status_and_redacted_provider_message(provider: bool) -> None: + import httpx + + from litellm.proxy._types import ProxyException + from litellm.proxy.lens.endpoints import model_failure + from litellm.proxy.lens.worker import failure_message + + message: Final = "Token rate limit exceeded. api_key=secret-example-value-123456 Retry in 60 seconds." + error: Final = model_failure( + ProxyException(message, "rate_limit_error", None, 429, headers={"retry-after": "60"}) + if provider + else HTTPException(429, message, headers={"retry-after": "60"}) + ) + request: Final = httpx.Request("POST", "https://proxy.test/lens/worker/lens/run/model") + response: Final = httpx.Response(error.status_code, json={"detail": error.detail}, request=request) + with pytest.raises(httpx.HTTPStatusError) as caught: + response.raise_for_status() + diagnostic: Final = failure_message(caught.value) + assert diagnostic.startswith("Model request failed (HTTP 429):") + assert "Token rate limit exceeded." in diagnostic + assert "Retry in 60 seconds." in diagnostic + assert "secret-example" not in diagnostic + assert error.headers == {"retry-after": "60"} + + +@pytest.mark.asyncio +async def test_preview_reports_calendar_overflow_as_a_validation_error() -> None: + from datetime import datetime, timezone + + from litellm.proxy.lens.endpoints import Preview, preview_sample + + body: Final = Preview( + settings=LensSettings(name="Calendar regression", model="analysis", context="Read recorded activity"), + as_of=datetime.min.replace(tzinfo=timezone.utc), + ) + with pytest.raises(HTTPException) as error: + await preview_sample(body, UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN), None) + assert error.value.status_code == 422 + assert "supported calendar range" in error.value.detail diff --git a/tests/unit/proxy/lens/test_inference.py b/tests/unit/proxy/lens/test_inference.py index 3b69d624a7e..e249693c4d5 100644 --- a/tests/unit/proxy/lens/test_inference.py +++ b/tests/unit/proxy/lens/test_inference.py @@ -15,6 +15,7 @@ def test_missing_optional_price_tiers_use_base_rates(monkeypatch: pytest.MonkeyP "openai/lens-base-rate-test": { "litellm_provider": "openai", "mode": "chat", + "max_output_tokens": 16384, "input_cost_per_token": 0.001, "output_cost_per_token": 0.002, "input_cost_per_token_above_200k_tokens": None, @@ -27,7 +28,10 @@ def test_missing_optional_price_tiers_use_base_rates(monkeypatch: pytest.MonkeyP deployment: Final = Deployment(litellm_params=DeploymentParams(model="openai/lens-base-rate-test")) explicit: Final = Deployment( litellm_params=DeploymentParams( - model="openai/lens-base-rate-test", input_cost_per_token=0.001, output_cost_per_token=0.002 + model="openai/lens-base-rate-test", + input_cost_per_token=0.001, + output_cost_per_token=0.002, + max_tokens=16384, ) ) assert quote((deployment,), "Answer the question") == quote((explicit,), "Answer the question") @@ -45,7 +49,7 @@ def test_unpriced_model_requires_explicit_rates() -> None: def test_custom_priced_model_charges_reported_tokens() -> None: deployment: Final = Deployment( litellm_params=DeploymentParams( - model="openai/lens-test", input_cost_per_token=0.001, output_cost_per_token=0.002 + model="openai/lens-test", input_cost_per_token=0.001, output_cost_per_token=0.002, max_tokens=16384 ) ) response: Final = ModelResponse( @@ -53,3 +57,66 @@ def test_custom_priced_model_charges_reported_tokens() -> None: ) assert completion_charge((deployment,), response, 10) == pytest.approx(0.04) assert quote((deployment,), "hello") > 0.04 + + +@pytest.mark.parametrize("capacity", (8192, 65536, 128000)) +def test_output_allowance_and_budget_follow_the_models_capacity(capacity: int, monkeypatch: pytest.MonkeyPatch) -> None: + from litellm.proxy.lens.inference import output_tokens + + monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost}) + litellm.register_model( + model_cost={ + "openai/lens-capacity-test": { + "litellm_provider": "openai", + "mode": "chat", + "max_output_tokens": capacity, + "input_cost_per_token": 0, + "output_cost_per_token": 0.001, + } + } + ) + deployment: Final = Deployment(litellm_params=DeploymentParams(model="openai/lens-capacity-test")) + assert output_tokens(deployment) == capacity + assert quote((deployment,), "Review") == pytest.approx(capacity * 0.001) + + +def test_explicit_deployment_output_setting_is_respected() -> None: + from litellm.proxy.lens.inference import output_tokens + + deployment: Final = Deployment(litellm_params=DeploymentParams(model="custom/model", max_tokens=32000)) + assert output_tokens(deployment) == 32000 + + +def test_shared_context_capacity_leaves_room_for_the_entire_prompt(monkeypatch: pytest.MonkeyPatch) -> None: + from litellm.proxy.lens.inference import output_tokens + + monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost}) + litellm.register_model( + model_cost={ + "openai/lens-shared-context": { + "litellm_provider": "openai", + "mode": "chat", + "max_output_tokens": 8192, + "max_input_tokens": 8192, + "input_cost_per_token": 0, + "output_cost_per_token": 0.001, + } + } + ) + deployment: Final = Deployment(litellm_params=DeploymentParams(model="openai/lens-shared-context")) + short: Final = output_tokens(deployment, "Review this trace") + long: Final = output_tokens(deployment, "Review this trace " * 500) + assert 0 < long < short < output_tokens(deployment) + assert quote((deployment,), "Review this trace " * 500) == pytest.approx(long * 0.001) + + +def test_unknown_model_capacity_requires_explicit_operator_metadata() -> None: + from litellm.proxy.lens.inference import ModelCapacity, output_tokens + + params: Final = DeploymentParams(model="openai/lens-unknown-capacity") + with pytest.raises(HTTPException) as error: + output_tokens(Deployment(litellm_params=params)) + assert error.value.status_code == 400 + assert "model_info.max_output_tokens" in error.value.detail + configured: Final = Deployment(litellm_params=params, model_info=ModelCapacity(max_output_tokens=32000)) + assert output_tokens(configured) == 32000 diff --git a/tests/unit/proxy/lens/test_sources.py b/tests/unit/proxy/lens/test_sources.py index bae06f4afac..81bae7a0091 100644 --- a/tests/unit/proxy/lens/test_sources.py +++ b/tests/unit/proxy/lens/test_sources.py @@ -57,7 +57,10 @@ async def test_sample_never_returns_authentication_attributes() -> None: reader: Final = SourceReader(StorageResponse()) sample: Final = await reader.sample(Scope(team_id="alpha"), lens().settings, 1, 2) - assert sample.executions[0].metadata == (MetadataFilter(key="environment", value="production"),) + assert sample.executions[0].metadata == ( + MetadataFilter(key="environment", value="production"), + MetadataFilter(key="oversized", value="x" * 501), + ) assert "opaque-oauth-bearer" not in sample.model_dump_json() assert sample.eligible == 1 diff --git a/tests/unit/proxy/lens/test_state.py b/tests/unit/proxy/lens/test_state.py index c4220b7dd6d..fac43d6e850 100644 --- a/tests/unit/proxy/lens/test_state.py +++ b/tests/unit/proxy/lens/test_state.py @@ -103,7 +103,6 @@ def test_behavior_description_is_sufficient_without_separate_checks() -> None: ("sample_size", 0), ("concurrency", 0), ("lookback_hours", 0), - ("lookback_hours", 8761), ), ) def test_invalid_selection_and_parallelism_are_rejected(field: str, value: int) -> None: @@ -217,7 +216,7 @@ def test_custom_schedule_does_not_overlap_an_active_scan(interval: int) -> None: assert queue_job(running, NOW + timedelta(minutes=interval), "second") is running -@pytest.mark.parametrize("interval", (0, -1, 10081, 1.5)) +@pytest.mark.parametrize("interval", (0, -1, 1.5)) def test_invalid_schedule_is_rejected(interval: float) -> None: from pydantic import ValidationError @@ -286,3 +285,13 @@ def test_legacy_finding_identity_preserves_feedback_only_for_same_kind_and_check separate: Final = merge_finding(reviewed, other, 2, NOW) assert separate.id != legacy_id assert separate.status == "open" and separate.reason == "" + + +@pytest.mark.parametrize("field", ("lookback_hours", "interval_minutes")) +def test_calendar_overflow_is_rejected_without_the_old_history_and_interval_caps(field: str) -> None: + from pydantic import ValidationError + + accepted: Final = LensSettings.model_validate({**lens().settings.model_dump(), field: 100000}) + assert getattr(accepted, field) == 100000 + with pytest.raises(ValidationError, match="supported calendar range"): + LensSettings.model_validate({**lens().settings.model_dump(), field: 10**30}) diff --git a/tests/unit/proxy/lens/test_worker.py b/tests/unit/proxy/lens/test_worker.py index 0e64b0f10c8..17bac711e39 100644 --- a/tests/unit/proxy/lens/test_worker.py +++ b/tests/unit/proxy/lens/test_worker.py @@ -1,3 +1,4 @@ +import asyncio from queue import SimpleQueue from typing import Final @@ -89,7 +90,8 @@ async def test_incompatible_claim_reports_failure_instead_of_leaving_the_investi ) -> None: claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) payload: Final = claim.model_dump(mode="json") | { - "job": claim.job.model_dump(mode="json") | { + "job": claim.job.model_dump(mode="json") + | { "settings": claim.job.settings.model_dump() | {"future_setting": "private content"}, }, } @@ -197,3 +199,221 @@ def test_connection_timeout_and_invalid_response_have_distinct_private_diagnosti assert "connect to the proxy" in failure_message(httpx.ConnectError("private hostname")) assert "timed out" in failure_message(httpx.ReadTimeout("private prompt")) assert "structured JSON" in failure_message(ValueError("private model response")) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "purpose,stage,schema", + ( + ("extract", "Reading executions", "TraceReview"), + ("cluster", "Grouping observations", "Clusters"), + ("investigate", "Checking original evidence", "Decision"), + ), +) +async def test_worker_saves_validation_errors_from_every_analysis_stage(purpose: str, stage: str, schema: str) -> None: + import json + + claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) + execution: Final = Execution( + id="run", source="traces", trace_id="trace", team_id="alpha", name="review", start_time="", span_count=1 + ) + sample: Final = Sample(executions=(execution,), eligible=1) + content: Final = ExecutionContent( + execution=execution, + parts=(TracePart(execution_id="run", span_id="span", name="lead", kind="agent", content="Tool timeout"),), + ) + saved: Final = SimpleQueue[Result]() + attempts: Final = SimpleQueue[str]() + + def handle(request: httpx.Request) -> httpx.Response: + match request.url.path.rsplit("/", 1)[-1]: + case "claim": + return httpx.Response(200, json=claim.model_dump(mode="json")) + case "sample": + return httpx.Response(200, json=sample.model_dump(mode="json")) + case "content": + return httpx.Response(200, json=content.model_dump(mode="json")) + case "model": + body: Final = ModelRequest.model_validate_json(request.content) + if body.purpose == purpose: + attempts.put(body.purpose) + return httpx.Response( + 200, + json={"content": '{"candidates":[', "cost": 0.01}, + headers={"x-litellm-lens-finish-reason": "length"}, + ) + if body.purpose == "cluster": + return httpx.Response( + 200, + json={ + "content": json.dumps({"candidates": json.loads(body.prompt)["candidates"]}), + "cost": 0.01, + }, + ) + return httpx.Response( + 200, + json={ + "content": json.dumps( + { + "observations": [ + { + "check_id": claim.job.settings.analysis_checks[0].id, + "summary": "Tool timeout", + "evidence": [ + {"execution_id": "r0", "span_id": "span", "quote": "Tool timeout"} + ], + } + ] + } + ), + "cost": 0.01, + }, + ) + case "progress": + return httpx.Response(200, json=True) + case "result": + saved.put(Result.model_validate_json(request.content)) + return httpx.Response(200, json=True) + case _: + pytest.fail(f"Unexpected worker request: {request.url.path}") + + async with httpx.AsyncClient(base_url="https://proxy.test", transport=httpx.MockTransport(handle)) as client: + assert await LensWorker(client).run_once() + message: Final = saved.get_nowait().error + assert message.startswith(f"{stage} failed: {schema} response invalid after 2 attempts.") + assert "finish_reason=length" in message + assert "EOF while parsing" in message and "[json_invalid]" in message + assert attempts.qsize() == 2 and saved.empty() + + +def test_response_validation_diagnostics_omit_input_values_and_unexpected_field_names() -> None: + with pytest.raises(ValidationError) as caught: + ModelResult.model_validate({"content": "private trace", "cost": "private token", "private field": "secret"}) + message: Final = failure_message(caught.value) + assert "Invalid ModelResult response" in message + assert "cost:" in message and "[float_parsing]" in message + assert "[extra_forbidden]" in message + assert "private" not in message and "secret" not in message + + +@pytest.mark.asyncio +@pytest.mark.parametrize("heartbeat_status", (401, 403, 409)) +async def test_losing_the_lease_interrupts_an_in_flight_model_request(heartbeat_status: int) -> None: + claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) + execution: Final = Execution( + id="run", source="traces", trace_id="t", team_id="", name="task", start_time="", span_count=1 + ) + started: Final = asyncio.Event() + cancelled: Final = asyncio.Event() + never: Final = asyncio.Event() + saved: Final = SimpleQueue[Result]() + + async def heartbeat_wait(_seconds: float) -> None: + await started.wait() + + async def handle(request: httpx.Request) -> httpx.Response: + match request.url.path.rsplit("/", 1)[-1]: + case "claim": + return httpx.Response(200, json=claim.model_dump(mode="json")) + case "sample": + return httpx.Response(200, json=Sample(executions=(execution,), eligible=1).model_dump()) + case "content": + return httpx.Response( + 200, + json=ExecutionContent( + execution=execution, + parts=( + TracePart(execution_id="run", span_id="span", name="step", kind="tool", content="evidence"), + ), + ).model_dump(), + ) + case "model": + assert request.extensions["timeout"] == {"connect": 13, "read": None, "write": 13, "pool": 13} + started.set() + try: + await never.wait() + finally: + cancelled.set() + pytest.fail("The cancelled model request must not finish") + case "heartbeat": + return httpx.Response(heartbeat_status) + case "progress": + return httpx.Response(200, json=True) + case "result": + saved.put(Result.model_validate_json(request.content)) + return httpx.Response(409) + case _: + pytest.fail(f"Unexpected worker request: {request.url.path}") + + async with httpx.AsyncClient( + base_url="https://proxy.test", transport=httpx.MockTransport(handle), timeout=13 + ) as client: + assert await LensWorker(client, heartbeat_wait=heartbeat_wait).run_once() + assert cancelled.is_set() + assert f"HTTP {heartbeat_status}" in saved.get_nowait().error + assert saved.empty() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", (429, 500, 502, 503, 504, "connection", "timeout")) +async def test_transient_heartbeat_failure_recovers_without_cancelling_analysis(failure: int | str) -> None: + claim: Final = Claim(lens_id="lens", job=queue_job(lens(), NOW, "job").jobs[0], findings=()) + execution: Final = Execution( + id="run", source="traces", trace_id="t", team_id="", name="task", start_time="", span_count=1 + ) + started: Final = asyncio.Event() + recovered: Final = asyncio.Event() + never: Final = asyncio.Event() + attempts: Final = SimpleQueue[str]() + saved: Final = SimpleQueue[Result]() + + async def heartbeat_wait(_seconds: float) -> None: + await started.wait() + if attempts.qsize() >= 2: + await never.wait() + + async def handle(request: httpx.Request) -> httpx.Response: + match request.url.path.rsplit("/", 1)[-1]: + case "claim": + return httpx.Response(200, json=claim.model_dump(mode="json")) + case "sample": + return httpx.Response(200, json=Sample(executions=(execution,), eligible=1).model_dump()) + case "content": + return httpx.Response( + 200, + json=ExecutionContent( + execution=execution, + parts=( + TracePart(execution_id="run", span_id="span", name="step", kind="tool", content="evidence"), + ), + ).model_dump(), + ) + case "model": + started.set() + await recovered.wait() + return httpx.Response(200, json={"content": '{"observations":[],"cannot_assess":false}', "cost": 0.01}) + case "heartbeat": + attempts.put(request.url.path) + if attempts.qsize() == 1: + if failure == "connection": + raise httpx.ConnectError("temporary connection failure", request=request) + if failure == "timeout": + raise httpx.ReadTimeout("temporary response timeout", request=request) + assert isinstance(failure, int) + return httpx.Response(failure) + recovered.set() + return httpx.Response(200, json=True) + case "progress": + return httpx.Response(200, json=True) + case "result": + saved.put(Result.model_validate_json(request.content)) + return httpx.Response(200, json=True) + case _: + pytest.fail(f"Unexpected worker request: {request.url.path}") + + async with httpx.AsyncClient(base_url="https://proxy.test", transport=httpx.MockTransport(handle)) as client: + assert await LensWorker(client, heartbeat_wait=heartbeat_wait).run_once() + result: Final = saved.get_nowait() + assert result.error == "" + assert result.coverage.screened == 1 and result.coverage.unassessable == 0 + assert attempts.qsize() == 2 and saved.empty() diff --git a/ui/litellm-dashboard/src/components/lens/investigations/InvestigationsView.integration.test.tsx b/ui/litellm-dashboard/src/components/lens/investigations/InvestigationsView.integration.test.tsx index 52bb4707bad..f7c9c1e6ae8 100644 --- a/ui/litellm-dashboard/src/components/lens/investigations/InvestigationsView.integration.test.tsx +++ b/ui/litellm-dashboard/src/components/lens/investigations/InvestigationsView.integration.test.tsx @@ -506,3 +506,23 @@ it("closes editing when browser navigation leaves the investigation", async () = await waitFor(() => expect(screen.queryByRole("dialog")).not.toBeInTheDocument()); expect(apiClient.request).not.toHaveBeenCalled(); }); + +it("shows the actual saved failure and run context without opening backend logs", async () => { + testQueryClient.clear(); + const error = + "Grouping observations failed: Clusters response invalid after 2 attempts.\n" + + "candidates.0.check_id: Field required [missing]"; + const job = { ...lens.jobs[0], id: "failed-run", status: "failed" as const, stage: "Failed", error, findings: [] }; + vi.mocked(apiClient.get).mockImplementation(async (path) => { + if (path === "/lens") return { lenses: [{ ...lens, jobs: [job] }], workers: [], tracing_enabled: true }; + if (path === "/lens/lens/runs") return [job]; + if (path === "/lens/lens/runs/failed-run") return job; + return { data: [] }; + }); + renderWithProviders(); + const failure = within(await screen.findByRole("alert")); + expect(failure.getByLabelText("Investigation error")).toHaveTextContent(error.replaceAll("\n", " ")); + expect(failure.getByText("failed-run")).toBeVisible(); + expect(failure.getByText(job.settings.model)).toBeVisible(); + expect(failure.queryByText(/find the error in proxy and worker logs/)).not.toBeInTheDocument(); +}); diff --git a/ui/litellm-dashboard/src/components/lens/investigations/detail/InvestigationFailure.tsx b/ui/litellm-dashboard/src/components/lens/investigations/detail/InvestigationFailure.tsx index 4fa906a1b7f..ac6dc65babb 100644 --- a/ui/litellm-dashboard/src/components/lens/investigations/detail/InvestigationFailure.tsx +++ b/ui/litellm-dashboard/src/components/lens/investigations/detail/InvestigationFailure.tsx @@ -17,9 +17,11 @@ export function InvestigationFailure({ job, connected, className, ...props }: In className={cn("space-y-2 rounded-md border border-destructive/20 p-3 text-sm", className)} >

This investigation did not finish

-

{job.error}

-
- Troubleshooting details + 
+        {job.error}
+
+
+ Run details
Run:
@@ -38,10 +40,6 @@ export function InvestigationFailure({ job, connected, className, ...props }: In
{runTime(job.created_at)}
-

- Use the run ID to find the error in proxy and worker logs. Check the worker key's model permissions and - budget before retrying. -

); diff --git a/ui/litellm-dashboard/src/components/lens/setup/MatchingActivity.tsx b/ui/litellm-dashboard/src/components/lens/setup/MatchingActivity.tsx index da0ca45860e..38971aa19af 100644 --- a/ui/litellm-dashboard/src/components/lens/setup/MatchingActivity.tsx +++ b/ui/litellm-dashboard/src/components/lens/setup/MatchingActivity.tsx @@ -50,7 +50,7 @@ export function MatchingActivity({ return () => clearTimeout(timer); }, [serialized]); const historyHours = selection.lookback_hours ?? 24; - const validWindow = Number.isInteger(historyHours) && historyHours >= 1 && historyHours <= 8760; + const validWindow = Number.isInteger(historyHours) && historyHours >= 1; const percent = scope.sample_percent ?? 100; const cap = scope.sample_size; const validCap = cap == null || (Number.isInteger(cap) && cap > 0); diff --git a/ui/litellm-dashboard/src/components/lens/setup/MonitoringDialog.tsx b/ui/litellm-dashboard/src/components/lens/setup/MonitoringDialog.tsx index b283673dcf6..9ddac5dd281 100644 --- a/ui/litellm-dashboard/src/components/lens/setup/MonitoringDialog.tsx +++ b/ui/litellm-dashboard/src/components/lens/setup/MonitoringDialog.tsx @@ -15,7 +15,7 @@ import { DurationInput } from "@/components/shared/DurationInput"; import { type Settings } from "../model/types"; const monitoringSchema = z.object({ - interval_minutes: z.number().int().min(1).max(10080), + interval_minutes: z.number().int().min(1), }); export function MonitoringDialog({ @@ -62,13 +62,7 @@ export function MonitoringDialog({ control={control} name="interval_minutes" render={({ field }) => ( - + )} /> {formState.errors.interval_minutes?.message && ( diff --git a/ui/litellm-dashboard/src/components/lens/setup/fields/SampleFields.tsx b/ui/litellm-dashboard/src/components/lens/setup/fields/SampleFields.tsx index 5af37dadca5..305a28af34b 100644 --- a/ui/litellm-dashboard/src/components/lens/setup/fields/SampleFields.tsx +++ b/ui/litellm-dashboard/src/components/lens/setup/fields/SampleFields.tsx @@ -18,13 +18,7 @@ export function SampleFields() { control={control} name="selection.lookback_hours" render={({ field }) => ( - + )} /> {errors.selection?.lookback_hours?.message && ( diff --git a/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.test.ts b/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.test.ts index 95e16a12a8a..7b034dcbe2c 100644 --- a/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.test.ts +++ b/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.test.ts @@ -38,11 +38,11 @@ describe("investigation validation", () => { ); }); - it.each([0, 0.5, 8761, NaN, Infinity])("rejects an invalid history window: %s", (lookback_hours) => { + it.each([0, 0.5, NaN, Infinity])("rejects an invalid history window: %s", (lookback_hours) => { expectIssue( { selection: { lookback_hours } }, ["selection", "lookback_hours"], - "Choose a time range between 1 hour and 365 days", + "Choose a time range of at least 1 hour", ); }); @@ -62,10 +62,13 @@ describe("investigation validation", () => { ); }); - it.each([1, 8760])("accepts the history boundary %s with fractional sampling and no maximum", (lookback_hours) => { - const result = parse({ selection: { lookback_hours, sample_percent: 0.01, sample_size: null } }); - expect(result.success).toBe(true); - }); + it.each([1, 8760, 8761, 100000])( + "accepts the history window %s with fractional sampling and no maximum", + (lookback_hours) => { + const result = parse({ selection: { lookback_hours, sample_percent: 0.01, sample_size: null } }); + expect(result.success).toBe(true); + }, + ); it("requires individual runs only on the Run step", () => { expectIssue( @@ -89,8 +92,8 @@ describe("investigation validation", () => { }); it("validates budget and repeat interval with field-specific paths", () => { - expectIssue({ budget: Infinity }, ["budget"], "Choose a monthly limit greater than zero and up to 100000"); - expectIssue({ repeat: true, interval: 1.5 }, ["interval"], "Choose a repeat interval between 1 and 10080 minutes"); + expectIssue({ budget: Infinity }, ["budget"], "Choose a monthly limit greater than zero"); + expectIssue({ repeat: true, interval: 1.5 }, ["interval"], "Choose a repeat interval of at least 1 minute"); expect(parse({ repeat: false, interval: 0 }).success).toBe(true); expect(parse({ repeat: true, interval: 10080 }).success).toBe(true); }); diff --git a/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.ts b/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.ts index 0db08dbb836..2d3efc9abd2 100644 --- a/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.ts +++ b/ui/litellm-dashboard/src/components/lens/setup/investigationSchema.ts @@ -64,10 +64,10 @@ function validateManualSelection(draft: InvestigationDraft, ctx: z.RefinementCtx function validateSampleWindow(draft: InvestigationDraft, ctx: z.RefinementCtx) { const selection = draft.selection; const hours = selection.lookback_hours ?? 24; - if (!Number.isInteger(hours) || hours < 1 || hours > 8760) { + if (!Number.isInteger(hours) || hours < 1) { ctx.addIssue({ code: "custom", - message: "Choose a time range between 1 hour and 365 days", + message: "Choose a time range of at least 1 hour", path: ["selection", "lookback_hours"], }); } @@ -89,19 +89,19 @@ function validateSampleWindow(draft: InvestigationDraft, ctx: z.RefinementCtx) { } function validateBudgetAndSchedule(draft: InvestigationDraft, ctx: z.RefinementCtx) { - if (!Number.isFinite(draft.budget) || draft.budget <= 0 || draft.budget > 100000) { + if (!Number.isFinite(draft.budget) || draft.budget <= 0) { ctx.addIssue({ code: "custom", - message: "Choose a monthly limit greater than zero and up to 100000", + message: "Choose a monthly limit greater than zero", path: ["budget"], }); } - const intervalOutOfRange = draft.interval < 1 || draft.interval > 10080; + const intervalOutOfRange = draft.interval < 1; const intervalInvalid = !Number.isInteger(draft.interval) || intervalOutOfRange; if (draft.repeat && intervalInvalid) { ctx.addIssue({ code: "custom", - message: "Choose a repeat interval between 1 and 10080 minutes", + message: "Choose a repeat interval of at least 1 minute", path: ["interval"], }); } @@ -208,7 +208,7 @@ export function investigationSettings( return { ...initial, ...draft.selection, - name: draft.name.trim() || suggestedName.slice(0, 100), + name: draft.name.trim() || suggestedName, context: draft.context, model, monthly_budget: draft.budget, diff --git a/ui/litellm-dashboard/src/components/lens/setup/steps/ExpectationsStep.tsx b/ui/litellm-dashboard/src/components/lens/setup/steps/ExpectationsStep.tsx index 3dc696aee8a..128d58b8126 100644 --- a/ui/litellm-dashboard/src/components/lens/setup/steps/ExpectationsStep.tsx +++ b/ui/litellm-dashboard/src/components/lens/setup/steps/ExpectationsStep.tsx @@ -22,7 +22,6 @@ export function ExpectationsStep() { What should the agent be doing?