tPjDDUdZddlmZddlZddlZddlZddlZddlmZm Z ddlm Z ddl m Z m Z ddl mZddlmZmZmZdd lmZdd lmZdd lmZer dd lmZdd lmZdZded<dZded<dZded<e ddGddZ GddeZ!ddZ"ddZ#ddZ$dZ%ded<dd!Z& dd#Z'e(d$ejRe DZ*ded%<dd&Z+e+d'Z,ded(<e+d)Z-ded*<e+d+Z.ded,<e+d-Z/ded.<e+d/Z0ded0<e+d1Z1ded2<e+d3d4Z2ded5<d6e2d7e2d8Z3ded9<e-d:Z4ded;<e4de,Z6ded?<e4d@e,dAe,Z7dedB<e7dCe.dDe/dEe,dFZ8dedG<dHe-dIe,dJZ9dedK<e1dLZ:dedM<dNjwgdOe+dPdQe+d dQe+dRdQe+dSdQe-dTe-dUe,dVe,dWe-dXe0dYe.dZe/d[e/d\e0d]ed^e0d]ed_e+d`dae+dbdce+d"dde3dee4dfe5dge5dCe/dhe6die6dCe/dje7dke7dCe/dle8dCe9dme,dne8dCe9doe,dpe,dqe:dre,dse-dte,due-dve4dwe0dxe,dye,dze0d{e4d|e,d}e2d~e+dSde2de2dZ ddZ? d ddZ@y)a Per-session auto-router benchmarks rollup. At request time the spend writer builds one AutoRouterTurnTransaction per successful auto-routed request (a request whose metadata carries a routing_decision) and queues it on the prisma client. The spend-log flush job drains the queue into LiteLLM_AutoRouterSession with one conditional upsert per turn: the statement classifies the turn (same model, first visit, return to a model the session already used, out of order) against the row's own columns, so nothing is read before the write and concurrent pods compose. The benchmarks endpoint aggregates these rows and never touches LiteLLM_SpendLogs. ) annotationsN)MappingSequence) dataclass)datetimetimezone)groupby) TYPE_CHECKINGFinal NamedTuple)verbose_proxy_logger)!INTERNAL_CALL_ORIGIN_METADATA_KEY)DB_RETRY_SAFE_ERROR_TYPES)SpendLogsPayload) PrismaClienti,r CACHE_TTL_5M_SECONDSiCACHE_TTL_1H_SECONDSa WITH windowed AS ( SELECT * FROM "LiteLLM_AutoRouterSession" WHERE last_turn_at >= $1::timestamp AND first_turn_at < $2::timestamp ), tier_maps AS ( SELECT router_name, router_type, jsonb_object_agg(tier, tier_turns) AS tier_turns FROM ( SELECT router_name, router_type, kv.key AS tier, SUM((kv.value)::int)::int AS tier_turns FROM windowed, LATERAL jsonb_each_text(tier_turns) AS kv GROUP BY router_name, router_type, kv.key ) per_tier GROUP BY router_name, router_type ) SELECT agg.*, COALESCE(tier_maps.tier_turns, '{}'::jsonb) AS tier_turns FROM ( SELECT router_name, router_type, COUNT(*)::int AS sessions, COALESCE(SUM(turns), 0)::int AS turns, COALESCE(SUM(unordered_turns), 0)::int AS unordered_turns, COALESCE(SUM(covered_turns), 0)::int AS covered_turns, COALESCE(SUM(cache_hits), 0)::int AS cache_hits, COALESCE(SUM(same_model_turns), 0)::int AS same_model_turns, COALESCE(SUM(same_model_hits), 0)::int AS same_model_hits, COALESCE(SUM(first_visit_turns), 0)::int AS first_visit_turns, COALESCE(SUM(first_visit_hits), 0)::int AS first_visit_hits, COALESCE(SUM(return_turns), 0)::int AS return_turns, COALESCE(SUM(return_hits), 0)::int AS return_hits, COALESCE(SUM(return_expired_misses), 0)::int AS return_expired_misses, COALESCE(SUM(return_within_ttl_misses), 0)::int AS return_within_ttl_misses, COALESCE(SUM(ttl_5m_turns), 0)::int AS ttl_5m_turns, COALESCE(SUM(ttl_1h_turns), 0)::int AS ttl_1h_turns, COALESCE(SUM(total_tokens), 0)::bigint AS total_tokens, COALESCE(SUM(spend), 0)::float8 AS spend, COALESCE(SUM(saved_spend), 0)::float8 AS saved_spend, COALESCE(SUM(EXTRACT(EPOCH FROM (last_turn_at - first_turn_at))), 0)::float8 AS session_seconds FROM windowed GROUP BY router_name, router_type ) agg LEFT JOIN tier_maps USING (router_name, router_type) ORDER BY agg.spend DESC AUTOROUTER_BENCHMARKS_SQLT)frozenslotsceZdZUded<ded<ded<ded<ded<ded<d ed <d ed <d ed <ded<ded<ded<ded<dZded<y)AutoRouterTurnTransactionstrapi_key session_id router_name router_typemodelrturn_atint total_tokensfloatspend saved_spendboolcovered cache_hit int | Nonecache_ttl_seconds cache_touchedNz str | Nonetier)__name__ __module__ __qualname____annotations__r+o/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/proxy/db/autorouter_session_rollup.pyrrUsT LO J  L MO!!D*r1rc:eZdZUdZded<ded<ded<ded<y ) TurnCacheFactsa/One statement of a turn's cache interaction, derived from its usage record. ``touched`` is False only when telemetry positively shows the provider neither read from nor wrote to the cache; absent telemetry reads as touched, which is the conservative input for the per-model idle clock. r%r&r read_tokensr(write_ttl_secondstouchedN)r,r-r.__doc__r/r0r1r2r4r4gsM!! Mr1r4cddlm}t|}||}t|}t |||| xs |dkDxs|duS)Nr)extract_cache_read_tokens)r&r5r6r7)$litellm.proxy.spend_tracking.savingsr:r%_write_ttl_secondsr4) usage_objectr:r&r5r6s r2turn_cache_factsr>usSN,'G2<@K1,? + O{QO2C42O  r1c tj|jdd}|j|S|j t jjdS#t$rYywxYw)NZz+00:00)tzinfo)r fromisoformatreplace ValueErrorrA astimezonerutc)start_time_isoparseds r2_turn_time_utcrIsl ..~/E/Ec8/TU}}   X\\ * 2 2$ 2 ?? s%A$$ A0/A0cddlm}|sy|jd}t|tr|jdnd}t|tr.|jdrt S|jdrt S||dkDrt Sy)zThe TTL this turn's cache write used, or None when nothing was written. Providers that report a TTL split do so under prompt_tokens_details; a write with no split is the provider's default five-minute cache. r)extract_cache_creation_tokensNprompt_tokens_detailscache_creation_token_detailsephemeral_1h_input_tokensephemeral_5m_input_tokens)r;rKget isinstancerrr)r=rKdetailscreations r2r<r<s S !%%&=>GEOPWY`Eagkk"@AgkH(G$ <<3 4' ' <<3 4' '$\2Q6## r1SESSION_ID_MAX_CHARSrct|tkr|Sdtj|j ddj zS)zThe session id as stored, bounded so a caller-chosen identifier cannot exceed Postgres's B-tree index entry limit through the composite primary key. Oversized ids map to a stable digest, so their turns still aggregate into one session.zsha256:zutf-8 surrogatepass)errors)lenrUhashlibsha256encode hexdigest)rs r2_bounded_session_idr^sD :.. w~~j&7&7&7&XYcce eer1r$c$|jddk7ry|jtry|jd}t|tr|sy|jdxs|jd}|jd}|jd}|jd }t|tr|r|r|r|syt t |jd xsd }|yd d lm} |jd} tt| tr| nd} |jd} | |} t|t||t |jdxsdt| tr| r| nd||t|jdxsd t|jdxsd zt|jdxsd| xsdz|| j| jd kD| j| j S)aOne rollup transaction for a successful auto-routed turn, else None. The routing_decision record is what says a request was auto-routed at all, so a request without one (including the auto-router's own classifier sub-calls) never reaches the rollup. Internal sub-calls that DO carry one (a shadow eval's duplicate of a request through the router) are excluded by their internal_call_origin stamp: they are not traffic a user sent, so counting them would manufacture sessions and savings in the adoption metrics. Failed requests served nothing and are excluded. The classifier's charge still lands here exactly once, via the decision's own classifier_cost folded into this turn's spend: the excluded classifier row is how it was billed, the decision is how it is attributed. Cache facts are derived from the payload's own usage record through the savings owner, never handed in beside it. statussuccessNrouting_decisionrouter_model_name model_grouprrr startTimer)classifier_cost_from_decisionr=r+runknown prompt_tokenscompletion_tokensr#g)rrrrr+rrr!r#r$r&r'r)r*)rPrrQrrrIr;rgr>rr^r r"r&r5r6r7)payloadmetadatar$rbrrrrrrgusage_object_rawcachetier_rawclassifier_costs r2!build_autorouter_turn_transactionrqs${{8 )||56&ll+=> & 08H)--.AB`gkkR_F`K[[+G L1J;;w'E {C ([WX]#C K(@(FB$GHGR&ll>:# CSU\8]$4cghE&**62H:;KLO $&z2(,,];HyI#Hc2xXT_5:;c'++NaBbBgfg>hhGKK(/C0O4JsK ##a'11mm r1c#4K|]}|jywN)name).0fields r2 rws#j5EJJ#js_UPSERT_PARAM_FIELDSc8dtj|dzS)zPositional placeholder for a transaction field, numbered by the dataclass's own field order so the SQL and the argument tuple cannot disagree; typos fail at import.$)rxindex) field_names r2_pr~s##))*59: ;;r1r_MODELr_TURN_ATr&_COVEREDr' _CACHE_HITr) _CACHE_TTLr*_TOUCHEDr+z::text_TIERz (CASE WHEN z2 IS NULL THEN '{}'::jsonb ELSE jsonb_build_object(z , 1) END) _TIER_DELTAz::timestamp >= t.last_turn_at _IN_ORDERz AND t.last_model = _SAMEz AND NOT t.models ? _FIRSTz AND t.models ? z AND t.last_model <> _RETURNz AND z::int = 1 AND z::int = 0 AND (t.models -> z ->> 'ttl') IS NOT NULL _RETURN_MISSzEXTRACT(EPOCH FROM z::timestamp) - (t.models -> z ->> 'at')::float8 _IDLE_SECONDSz ::int = 1_CACHE_TOUCHEDrfa INSERT INTO "LiteLLM_AutoRouterSession" AS t ( api_key, session_id, router_name, router_type, first_turn_at, last_turn_at, last_model, models, turns, unordered_turns, covered_turns, cache_hits, same_model_turns, same_model_hits, first_visit_turns, first_visit_hits, return_turns, return_hits, return_expired_misses, return_within_ttl_misses, ttl_5m_turns, ttl_1h_turns, total_tokens, spend, saved_spend, tier_turns ) VALUES ( rz, rrz ::timestamp, z::timestamp, z, jsonb_build_object(z., jsonb_build_object('at', EXTRACT(EPOCH FROM z::timestamp), 'ttl', z::int)), 1, 0, z::int, z::int, 0, 0, 1, z&::int, 0, 0, 0, 0, (CASE WHEN z::int = z$ THEN 1 ELSE 0 END), (CASE WHEN z THEN 1 ELSE 0 END), r!z ::bigint, r#z ::float8, z::float8, a ) ON CONFLICT (api_key, session_id, router_name) DO UPDATE SET turns = t.turns + 1, total_tokens = t.total_tokens + EXCLUDED.total_tokens, spend = t.spend + EXCLUDED.spend, saved_spend = t.saved_spend + EXCLUDED.saved_spend, covered_turns = t.covered_turns + EXCLUDED.covered_turns, cache_hits = t.cache_hits + EXCLUDED.cache_hits, ttl_5m_turns = t.ttl_5m_turns + EXCLUDED.ttl_5m_turns, ttl_1h_turns = t.ttl_1h_turns + EXCLUDED.ttl_1h_turns, unordered_turns = t.unordered_turns + (CASE WHEN NOT (zM) THEN 1 ELSE 0 END), same_model_turns = t.same_model_turns + (CASE WHEN zJ THEN 1 ELSE 0 END), same_model_hits = t.same_model_hits + (CASE WHEN zW::int = 1 THEN 1 ELSE 0 END), first_visit_turns = t.first_visit_turns + (CASE WHEN zL THEN 1 ELSE 0 END), first_visit_hits = t.first_visit_hits + (CASE WHEN zM::int = 1 THEN 1 ELSE 0 END), return_turns = t.return_turns + (CASE WHEN zB THEN 1 ELSE 0 END), return_hits = t.return_hits + (CASE WHEN zg::int = 1 THEN 1 ELSE 0 END), return_expired_misses = t.return_expired_misses + (CASE WHEN z > (t.models -> zw ->> 'ttl')::float8 THEN 1 ELSE 0 END), return_within_ttl_misses = t.return_within_ttl_misses + (CASE WHEN z <= (t.models -> zT ->> 'ttl')::float8 THEN 1 ELSE 0 END), models = t.models || jsonb_build_object(z/, jsonb_build_object( 'at', (CASE WHEN z4 THEN GREATEST(COALESCE((t.models -> z+ ->> 'at')::float8, 0), EXTRACT(EPOCH FROM z8::timestamp)) ELSE COALESCE((t.models -> z' ->> 'at')::float8, EXTRACT(EPOCH FROM z.::timestamp)) END), 'ttl', (CASE WHEN z THEN COALESCE(z::int, (t.models -> z= ->> 'ttl')::int) ELSE COALESCE((t.models -> z ->> 'ttl')::int, z0::int) END) )), last_model = (CASE WHEN z THEN z5 ELSE t.last_model END), tier_turns = (CASE WHEN z! IS NOT NULL AND t.router_type = z1 THEN t.tier_turns || jsonb_build_object(z, COALESCE((t.tier_turns ->> z)::int, 0) + 1) ELSE t.tier_turns END), first_turn_at = LEAST(t.first_turn_at, EXCLUDED.first_turn_at), last_turn_at = GREATEST(t.last_turn_at, EXCLUDED.last_turn_at) UPSERT_AUTOROUTER_SESSION_SQLc|t|tr t|St|tr|j S|Srs)rQr%r r isoformat)values r2 _as_sql_paramr=s2%5z%"  Lr1c4tfdtDS)Nc3HK|]}tt|ywrs)rgetattr)rurt transactions r2rwz!_upsert_params..Fs\tw{D9:\s")tuplerxrs`r2_upsert_paramsrEs \G[\ \\r1c 8Kt|dzD]6} |jjtgt |d{yy7#t $rA||k\rt jd|ztjddzd{7YwxYww)Nr{r) rangedb execute_rawrrrasynciosleeprandomuniform) prisma_clientr n_retry_timesattempts r2_upsert_turn_with_retryrIs *+ .-""../Lk~^iOjk k k  k( C-'--7 V^^Aq-A AB B B CsGB-A A A B A  ABBBBBBc VK|syt|d}t|dD]9\}}t|}t|D]\}} t |||d{;y7 #t $r<} t jd|dt||z t|| Yd} ~ d} ~ wwxYww)aDrain a queue batch into the rollup, one upsert per turn. Statements run sequentially in per-session event order: a turn's classification depends on the turns before it, and Postgres rejects one multi-row INSERT touching the same key twice. Only ConnectError is retried, per statement, because it proves that statement never reached the database. Any other failure drops the remaining turns of THAT session only, with an error log, and the flush continues with the next session: sessions are independent state machines, so one poisoned statement must not discard unrelated sessions, and a repeated increment is worse than an undercount. Callers must not add their own retry around this function. Nc^|j|j|j|jfSrs)rrrrrs r2z4flush_autorouter_turn_transactions..ms-     " "  # #    ! r1)keycH|j|j|jfSrs)rrrrs r2rz4flush_autorouter_turn_transactions..vs!4!4k6L6LkNeNe fr1z~Spend tracking - auto-router session rollup flush failed for router %s; %s of %s turn transactions dropped for one session: %sr) sortedr r enumerater Exceptionr errorrY) r transactionsrordered session_key session_group session_turnspositionr flush_errs r2"flush_autorouter_turn_transactionsrYs  G'. f'" ]m, %.}%=  !Hk -m[-XXX  Y $**MN &1 &  sBAB)A!AA!B)A!! B&*1B!B)!B&&B))r=Mapping[str, object] | Nonereturnr4)rGrrzdatetime | None)r=rrr()rrrr)rkrrlzMapping[str, object]r$r"rz AutoRouterTurnTransaction | None)r}rrr)rz$str | float | bool | datetime | Nonerzstr | float | None)rrrztuple[str | float | None, ...])rrrrrr rNone))rrrz#Sequence[AutoRouterTurnTransaction]rr rr)Ar8 __future__rr dataclassesrZrcollections.abcrrrrr itertoolsr typingr r r litellm._loggingr litellm.constantsrlitellm.proxy._typesrrlitellm.proxy.utilsrrr/rrrr4r>rIr<rUr^rqrfieldsrxr~rrrrrrrrrrrrrrrjoinrrrrrr0r1r2rs] # -!'331?:50!e!"e"-$5-` $d#$" Z  @,"e!f7 7"77& 7t$#jQ5(RT5(U]S]5(^k5(ltjt5(u5( H5( "5(#)5(*X5(YaWa5(bw5(xBwC5(C 5( *5(5('<5((5(\5(5(|5($5(%9#95(:5(|5($5(%9#95(:5( !5( #!5( $&g;-!5( 0:!5( ;=]:K9L!5( M!5("M#5(" ;#5(8w;5(:?6;5(<7;A?5(>B8?5(@9?xA5(@@EA5(@FPLA5(@Q0A5(B18yC5(B9.C5(D/6YE5(D7