scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(lens): preserve full trace access and expose investigation failures (#44406)

Browse source 
* fix(lens): preserve full trace access and expose investigation failures

* chore(lens): sync worker registration schema

* fix(lens): support durations without a configured maximum

* fix(lens): expose every page of fetched investigation evidence

* fix(lens): preserve repeated trace content and interrupt cancelled runs

* fix(lens): retry transient heartbeat failures during analysis
This commit is contained in:
moe-berri 2026-10-03 11:52:55 -07:00 • committed by GitHub
parent 50190134c3
commit af36e5c693
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
29 changed files with 1061 additions and 209 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -34909,7 +34909,6 @@
},
"name": {
"default": "Lens worker",
"maxLength": 100,
"minLength": 1,
"title": "Name",
"type": "string"

132
litellm/proxy/lens/analysis.py
View file

@ -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))),
)

31
litellm/proxy/lens/billing.py
View file

@ -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:

44
litellm/proxy/lens/endpoints.py
View file

@ -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)

130
litellm/proxy/lens/inference.py
View file

@ -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:

87
litellm/proxy/lens/models.py
View file

@ -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)

12
litellm/proxy/lens/prompts/investigate.md
View file

@ -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.

1
litellm/proxy/lens/prompts/review.md
View file

@ -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.

2
litellm/proxy/lens/sources.py
View file

@ -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

74
litellm/proxy/lens/worker.py
View file

@ -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

61
tests/proxy_behavior/lens/test_lifecycle.py
View file

@ -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)

226
tests/unit/proxy/lens/test_analysis.py
View file

@ -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")

42
tests/unit/proxy/lens/test_endpoints.py
View file

@ -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

71
tests/unit/proxy/lens/test_inference.py
View file

@ -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

5
tests/unit/proxy/lens/test_sources.py
View file

@ -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

13
tests/unit/proxy/lens/test_state.py
View file

@ -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})

222
tests/unit/proxy/lens/test_worker.py
View file

@ -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()

20
ui/litellm-dashboard/src/components/lens/investigations/InvestigationsView.integration.test.tsx
View file

@ -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(<InvestigationsView accessToken="test" readOnly />);
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();
});

12
ui/litellm-dashboard/src/components/lens/investigations/detail/InvestigationFailure.tsx
View file

@ -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)}
>
<p className="font-medium text-destructive">This investigation did not finish</p>
<p>{job.error}</p>
<details>
<summary className="cursor-pointer text-xs text-muted-foreground">Troubleshooting details</summary>
<pre className="whitespace-pre-wrap break-words font-mono text-xs" aria-label="Investigation error">
{job.error}
</pre>
<details open>
<summary className="cursor-pointer text-xs text-muted-foreground">Run details</summary>
<dl className="mt-2 space-y-1 text-xs text-muted-foreground">
<div>
<dt className="inline">Run: </dt>
@ -38,10 +40,6 @@ export function InvestigationFailure({ job, connected, className, ...props }: In
<dd className="inline">{runTime(job.created_at)}</dd>
</div>
</dl>
<p className="mt-2 text-xs text-muted-foreground">
Use the run ID to find the error in proxy and worker logs. Check the worker key&apos;s model permissions and
budget before retrying.
</p>
</details>
</div>
);

2
ui/litellm-dashboard/src/components/lens/setup/MatchingActivity.tsx
View file

@ -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);

10
ui/litellm-dashboard/src/components/lens/setup/MonitoringDialog.tsx
View file

@ -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 }) => (
<DurationInput
label="Check every"
value={field.value}
onChange={field.onChange}
base="minutes"
max={10080}
/>
<DurationInput label="Check every" value={field.value} onChange={field.onChange} base="minutes" />
)}
/>
{formState.errors.interval_minutes?.message && (

8
ui/litellm-dashboard/src/components/lens/setup/fields/SampleFields.tsx
View file

@ -18,13 +18,7 @@ export function SampleFields() {
control={control}
name="selection.lookback_hours"
render={({ field }) => (
<DurationInput
label="Review the last"
value={field.value ?? 24}
base="hours"
max={8760}
onChange={field.onChange}
/>
<DurationInput label="Review the last" value={field.value ?? 24} base="hours" onChange={field.onChange} />
)}
/>
{errors.selection?.lookback_hours?.message && (

19
ui/litellm-dashboard/src/components/lens/setup/investigationSchema.test.ts
View file

@ -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);
});

14
ui/litellm-dashboard/src/components/lens/setup/investigationSchema.ts
View file

@ -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,

1
ui/litellm-dashboard/src/components/lens/setup/steps/ExpectationsStep.tsx
View file

@ -22,7 +22,6 @@ export function ExpectationsStep() {
What should the agent be doing?
<Textarea
{...register("context")}
maxLength={6000}
rows={4}
placeholder="Answer the customer's question using verified sources and explain when information is missing."
/>

10
ui/litellm-dashboard/src/components/lens/setup/steps/RunStep.tsx
View file

@ -80,7 +80,7 @@ export function RunStep({
<div className="grid gap-5 sm:grid-cols-2">
<label className="grid content-start gap-2 text-sm font-medium">
Monthly limit (USD)
<Input {...register("budget", { valueAsNumber: true })} type="number" min="0.01" max="100000" step="1" />
<Input {...register("budget", { valueAsNumber: true })} type="number" min="0.01" step="1" />
{errors.budget?.message && (
<p role="alert" className="text-sm text-destructive">
{errors.budget.message}
@ -101,13 +101,7 @@ export function RunStep({
control={control}
name="interval"
render={({ field }) => (
<DurationInput
label="Repeat every"
value={field.value}
onChange={field.onChange}
base="minutes"
max={10080}
/>
<DurationInput label="Repeat every" value={field.value} onChange={field.onChange} base="minutes" />
)}
/>
)}

2
ui/litellm-dashboard/src/components/lens/setup/steps/ScopeStep.tsx
View file

@ -26,7 +26,7 @@ export function ScopeStep({
step === 0 ? (
<label className="grid gap-2 text-sm font-medium">
Investigation name
<Input {...register("name")} placeholder="e.g. Support quality" maxLength={100} />
<Input {...register("name")} placeholder="e.g. Support quality" />
</label>
) : undefined
}

14
ui/litellm-dashboard/src/components/shared/DurationInput.integration.test.tsx
View file

@ -3,17 +3,27 @@ import { useState } from "react";
import { describe, expect, it } from "vitest";
import { DurationInput } from "@/components/shared/DurationInput";
function DurationForm({ base, initial }: { base: "minutes" | "hours"; initial: number }) {
function DurationForm({ base, initial, max }: { base: "minutes" | "hours"; initial: number; max?: number }) {
const [value, setValue] = useState(initial);
return (
<>
<DurationInput label="Duration" base={base} value={value} onChange={setValue} max={10080} />
<DurationInput label="Duration" base={base} value={value} onChange={setValue} max={max} />
<output aria-label="Saved duration">{value}</output>
</>
);
}
describe("Duration units", () => {
it.each([undefined, 10080])("validates large durations against an optional maximum of %s", (max) => {
render(<DurationForm base="hours" initial={1} max={max} />);
fireEvent.change(screen.getByRole("spinbutton", { name: "Duration" }), { target: { value: 240000 } });
expect(screen.getByLabelText("Saved duration")).toHaveTextContent("240000");
if (max === undefined) {
expect(screen.getByRole("spinbutton", { name: "Duration" })).toBeValid();
} else {
expect(screen.getByRole("spinbutton", { name: "Duration" })).toBeInvalid();
}
});
it.each([
{ base: "hours" as const, initial: 24, unit: "1", displayed: 24 },
{ base: "minutes" as const, initial: 60, unit: "1", displayed: 60 },

4
ui/litellm-dashboard/src/components/shared/DurationInput.tsx
View file

@ -11,7 +11,7 @@ export type DurationInputProps = Omit<ComponentProps<"div">, "onChange"> & {
value: number;
onChange: (value: number) => void;
base: "minutes" | "hours";
max: number;
max?: number;
};
export function DurationInput({ label, value, onChange, base, max, className, ...props }: DurationInputProps) {
@ -41,7 +41,7 @@ export function DurationInput({ label, value, onChange, base, max, className, ..
id={id}
type="number"
min={1 / scale}
max={max / scale}
max={max === undefined ? undefined : max / scale}
step={1 / scale}
value={Number.isFinite(value) ? value / scale : ""}
onChange={(event) => onChange(event.target.value === "" ? NaN : Number(event.target.value) * scale)}