Ë
    uPêjc!  ã                  ó|   — d Z ddlmZ ddlZddlmZmZ ddlmZ ddl	m
Z
mZ eeeef   Zeeeef   Z G d„ d«      Zy)	aA  
In-memory queues for adaptive router state and session updates.

Pattern follows DailySpendUpdateQueue: hot path is fully in-memory; a background
flusher task drains the aggregator and writes batches to Postgres.

Two logical queues (one class):
  1. STATE updates: increments to (router, request_type, model) bandit cell.
     Aggregator key = (router_name, request_type, model_name)
     Aggregated payload = {"delta_alpha": float, "delta_beta": float, "samples_added": int}
  2. SESSION updates: full snapshot of a session row (last-write-wins per session+router+model).
     Aggregator key = (session_id, router_name, model_name)
     Aggregated payload = the full session state dict.

Hot-path API is non-blocking and synchronous from the caller's POV (it just appends
to the in-memory aggregator). Flush is async and batched.
é    )ÚannotationsN)ÚAnyÚFinal)Úverbose_router_logger)ÚAdaptiveRouterSessionRepositoryÚAdaptiveRouterStateRepositoryc                  ól   — e Zd ZdZd	d„Z	 	 	 	 	 	 	 	 	 	 	 	 d
d„Z	 	 	 	 	 	 	 	 	 	 dd„Zdd„Zdd„Zdd„Z	y)ÚAdaptiveRouterUpdateQueuez²
    Single class managing both state-update aggregation and session-snapshot aggregation.
    Held by the AdaptiveRouter strategy instance and started by the proxy on boot.
    c                ón   — i | _         i | _        t        j                  «       | _        d| _        d| _        y )Nr   )Ú
_state_aggÚ_session_aggÚasyncioÚLockÚ_lockÚ_max_state_size_seenÚ_max_session_size_seen©Úselfs    úy/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/router_strategy/adaptive_router/update_queue.pyÚ__init__z"AdaptiveRouterUpdateQueue.__init__(   s.   € Ø<>ˆŒØ>@ˆÔÜ—\‘\“^ˆŒ
Ø$%ˆÔ!Ø&'ˆÕ#ó    c              ƒ  óž  K  — |||f}| j                   4 ƒd{  –—†  | j                  j                  |«      }|€||ddœ| j                  |<   n'|dxx   |z  cc<   |dxx   |z  cc<   |dxx   dz  cc<   t        | j                  t        | j                  «      «      | _        ddd«      ƒd{  –—†  y7 Œ›7 Œ# 1 ƒd{  –—†7  sw Y   yxY w­w)zDAggregate a bandit-cell delta. Multiple deltas to the same cell sum.Né   )Údelta_alphaÚ
delta_betaÚsamples_addedr   r   r   )r   r   ÚgetÚmaxr   Úlen)r   Úrouter_nameÚrequest_typeÚ
model_namer   r   ÚkeyÚcurrents           r   Úadd_state_deltaz)AdaptiveRouterUpdateQueue.add_state_delta1   sÕ   è ø€ ð !,¨\¸:ÐFˆØ—:‘:÷ 	]ñ 	]Ø!Ÿ_™_×0Ñ0°Ó5ˆGØˆà#.Ø",Ø%&ñ(�—‘ Ò$ð ˜Ó&¨+Ñ5Ó&Ø˜Ó%¨Ñ3Ó%Ø˜Ó(¨AÑ-Ó(Ü(+¨D×,EÑ,EÄsÈ4Ï?É?ÓG[Ó(\ˆDÔ%÷	]÷ 	]ñ 	]øð 	]ø÷ 	]÷ 	]ñ 	]üsE   ‚C˜B4™CœBB8Â#CÂ.B6Â/CÂ6CÂ8C
Â>CÂ?C
ÃCc              ƒ  ó  K  — |||f}| j                   4 ƒd{  –—†  || j                  |<   t        | j                  t	        | j                  «      «      | _        ddd«      ƒd{  –—†  y7 ŒR7 Œ# 1 ƒd{  –—†7  sw Y   yxY w­w)zÛ
        Last-write-wins per session row. The state_dict is a snapshot of the
        SessionState (signals counts + bookkeeping fields). The flusher will
        upsert this into LiteLLM_AdaptiveRouterSession.
        N)r   r   r   r   r   )r   Ú
session_idr    r"   Ú
state_dictr#   s         r   Úadd_session_statez+AdaptiveRouterUpdateQueue.add_session_stateK   sƒ   è ø€ ð #-¨k¸:Ð!FˆØ—:‘:÷ 	cñ 	cØ%/ˆD×Ñ˜cÑ"Ü*-¨d×.IÑ.IÌ3Èt×O`ÑO`ÓKaÓ*bˆDÔ'÷	c÷ 	cñ 	cøð 	cø÷ 	c÷ 	cñ 	cüsD   ‚B˜A+™Bœ>A/ÁBÁ%A-Á&BÁ-BÁ/BÁ5A8Á6BÁ=Bc              ƒ  óR  K  — | j                   4 ƒd{  –—†  | j                  }i | _        ddd«      ƒd{  –—†  syt        |j                  «       «      D ]z  }|\  }}}||   }	 t	        |«      j
                  j                  d|||dœi||||d   |d   t        |d   «      dœd	|d   id	|d   id	t        |d   «      id
œdœ¬«      ƒ d{  –—†  Œ| t        |«      S 7 ŒË7 Œª# 1 ƒd{  –—†7  sw Y   ŒºxY w7 Œ*# t        $ r!}t        j                  d||«       Y d}~ŒÈd}~ww xY w­w)z{
        Drain state aggregator and apply to LiteLLM_AdaptiveRouterState.
        Returns number of cells flushed.
        Nr   Ú#router_name_request_type_model_name)r    r!   r"   r   r   r   )r    r!   r"   ÚalphaÚbetaÚtotal_samplesÚ	increment)r,   r-   r.   ©ÚcreateÚupdate©ÚwhereÚdataz;AdaptiveRouterUpdateQueue: failed to flush state for %s: %s)r   r   ÚsortedÚkeysr   ÚtableÚupsertÚintÚ	Exceptionr   Ú	exceptionr   )	r   Úprisma_clientÚbatchr#   ÚrouterÚrtÚmodelÚpayloadÚes	            r   Úflush_state_to_dbz+AdaptiveRouterUpdateQueue.flush_state_to_db^   st  è ø€ ð
 —:‘:÷ 	!ñ 	!ØŸ?™?ˆEØ ˆDŒO÷	!÷ 	!ñ Øô ˜%Ÿ*™*›,Ó'ò &	ˆCØ #ÑˆF�B˜Ø˜C‘jˆGð#ô 4°MÓB×HÑH×OÑOà=Ø+1Ø,.Ø*/ñ@ðð ,2Ø,.Ø*/Ø%,¨]Ñ%;Ø$+¨LÑ$9Ü-0°¸Ñ1IÓ-Jñ#ð '2°7¸=Ñ3IÐ%JØ%0°'¸,Ñ2GÐ$HØ.9¼3¸wÀÑ?WÓ;XÐ-Yñ#ñð Pó ÷ ñ ð&	ôP �5‹zÐðc	!øð 	!ø÷ 	!÷ 	!ñ 	!úð$ùô0 ò Ü%×/Ñ/ØQØØ÷ñ ûðüs~   ‚D'“C”D'—C#«D'¶C!·.D'Á&A'C:ÃC8ÃC:ÃD'Ã!D'Ã#C5Ã)C,Ã*C5Ã1D'Ã8C:Ã:	D$ÄDÄD'ÄD$Ä$D'c              ƒ  óP  K  — | j                   4 ƒd{  –—†  | j                  }i | _        ddd«      ƒd{  –—†  syt        |j                  «       «      D ]s  }|\  }}}||   }	 |j	                  «       D ��	ci c]  \  }}	|dvsŒ||	“Œ }
}}	t        |«      j                  j                  d|||dœi|||dœ|
¥|
dœ¬«      ƒ d{  –—†  Œu t        |«      S 7 ŒÄ7 Œ£# 1 ƒd{  –—†7  sw Y   Œ³xY wc c}	}w 7 Œ0# t        $ r!}t        j                  d||«       Y d}~ŒÇd}~ww xY w­w)z‰
        Drain session aggregator and upsert into LiteLLM_AdaptiveRouterSession.
        Returns number of session rows flushed.
        Nr   )r'   r    r"   Ú!session_id_router_name_model_namer0   r3   z=AdaptiveRouterUpdateQueue: failed to flush session for %s: %s)r   r   r6   r7   Úitemsr   r8   r9   r;   r   r<   r   )r   r=   r>   r#   r'   r?   rA   rB   ÚkÚvÚupdate_payloadrC   s               r   Úflush_session_to_dbz-AdaptiveRouterUpdateQueue.flush_session_to_db–   si  è ø€ ð
 —:‘:÷ 	#ñ 	#Ø×,Ñ,ˆEØ "ˆDÔ÷	#÷ 	#ñ Øä˜%Ÿ*™*›,Ó'ò $	ˆCØ(+Ñ%ˆJ˜ Ø˜C‘jˆGð!ð &-§]¡]£_÷"Ù!˜Q ¸ÐAlÒ8l�A�q‘Dð"�ñ "ô 6°mÓD×JÑJ×QÑQà;Ø*4Ø+1Ø*/ñ>ðð +5Ø+1Ø*/ñ#ð -ð	#ð #1ñð Ró ÷ ñ ð$	ôL �5‹zÐð[	#øð 	#ø÷ 	#÷ 	#ñ 	#üó""ðùô$ ò Ü%×/Ñ/ØSØØ÷ñ ûðüs�   ‚D&“C”D&—C«D&¶C·.D&Á&C9Á9C1ÂC1Â;C9ÃC7ÃC9ÃD&ÃD&ÃC.Ã"C%Ã#C.Ã*D&Ã1C9Ã9	D#ÄDÄD&ÄD#Ä#D&c              ƒ  ó
  K  — | j                   4 ƒd {  –—†  t        | j                  «      t        | j                  «      | j                  | j
                  dœcd d d «      ƒd {  –—†  S 7 ŒV7 Œ# 1 ƒd {  –—†7  sw Y   y xY w­w)N)Ústate_pendingÚsession_pendingÚmax_state_seenÚmax_session_seen)r   r   r   r   r   r   r   s    r   Ú
queue_sizez$AdaptiveRouterUpdateQueue.queue_sizeÌ   sm   è ø€ Ø—:‘:÷ 	ñ 	ä!$ T§_¡_Ó!5Ü#& t×'8Ñ'8Ó#9Ø"&×";Ñ";Ø$(×$?Ñ$?ñ	÷	÷ 	ò 	øð 	ø÷ 	÷ 	ñ 	üsE   ‚B“A*”B—AA.ÁBÁ$A,Á%BÁ,BÁ.B Á4A7Á5B Á<BN)ÚreturnÚNone)r    Ústrr!   rT   r"   rT   r   Úfloatr   rU   rR   rS   )
r'   rT   r    rT   r"   rT   r(   zdict[str, Any]rR   rS   )r=   r   rR   r:   )rR   zdict[str, int])
Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r%   r)   rD   rK   rQ   © r   r   r
   r
   "   sž   „ ñó
(ð]àð]ð ð]ð ð	]ð
 ð]ð ð]ð 
ó]ð4càðcð ðcð ð	cð
 #ðcð 
ócó&6óp2ôlr   r
   )rY   Ú
__future__r   r   Útypingr   r   Úlitellm._loggingr   Ú'litellm.repositories.table_repositoriesr   r   ÚtuplerT   ÚStateKeyÚ
SessionKeyr
   rZ   r   r   ú<module>rb      sK   ðñõ$ #ã ß å 2÷ð
 ��c˜3�Ñ€Ø�3˜˜S�=Ñ!€
÷qò qr   