o ưi@sUddlZddlZddlZddlZddlmZddlmZmZmZm Z ddl Z ddl Z ddl m Z mZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd l m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'ddl(m)Z)ddl*m+Z+m,Z,GdddZ-ddZ.Gddde-Z/Gddde-Z0Gddde-Z1ddl2m3Z3dd lmZ4gdZ5GdddZ6e$j7e$j8BZ9e:e;d<e:hdZrrWrQr_r`rrfrPrSr6r6r6r7r&s<  . p'>rcs(t|dd}|dur |S||IdHS)zg Module-level helper for tests to ensure hooks can be invoked even if the iterator is wrapped. r_N)r-)iteratorrTZhook_fnr6r6r7r`hs  r`cseZdZdZ    ddejdededede e ee fde ed e e ee fd e effd d Z d dZ defddZddZZS)ResponsesAPIStreamingIteratorzS Async iterator for processing streaming responses from the Responses API. Nrrrrr r!r"r#c (t||||||||||_dSN)superr8Z aiter_linesstream_iterator r5rrrrr r!r"r# __class__r6r7r8w  z&ResponsesAPIStreamingIterator.__init__cC|Srxr6rXr6r6r7 __aiter__z'ResponsesAPIStreamingIterator.__aiter__r9c sz;| z |jIdH}Wn tyd|_tw|||}|jr,t|dur<|j|dIdH}|WSqtyDtjyZ}z d|_| ||d}~wt yo}z d|_| ||d}~ww)NT)rT) r>rz __anext__StopAsyncIterationr+rWr_rr HTTPErrorrSrPr5rTrErVr6r6r7rsB      z'ResponsesAPIStreamingIterator.__anext__cCs|j}|jdur%t|jdr%z t|j|j}Wn ty$Ynwt|jj ||j t ddt j|jj|d|j t d|jt ddS)z7Handle logging for completed responses in async contextN model_dump)rEr$ra cache_hitrErr$rara)r,r\r@model_validaterrPasyncio create_taskrasync_success_handlerr$rr.rrksuccess_handlerrfr5Zlogging_responser6r6r7rQs2   z@ResponsesAPIStreamingIterator._handle_logging_completed_responserl)rnrorprqrrrsrLrrtrrrr8rrrrQ __classcell__r6r6r|r7rvrs4  &rvcseZdZdZ    ddejdededede e ee fde ed e e ee fd e effd d Z d dZ ddZddZZS)!SyncResponsesAPIStreamingIteratorzY Synchronous iterator for processing streaming responses from the Responses API. Nrrrrr r!r"r#c rwrx)ryr8 iter_linesrzr{r|r6r7r8r~z*SyncResponsesAPIStreamingIterator.__init__cCrrxr6rXr6r6r7__iter__rz*SyncResponsesAPIStreamingIterator.__iter__c Csz6| zt|j}Wn tyd|_tw|||}|jr(t|dur6t|j|d}|WSqty>tj yT}z d|_| ||d}~wt yi}z d|_| ||d}~ww)NT)rcrT) r>nextrz StopIterationr+rWr r_rrrrSrPrr6r6r7__next__sB     z*SyncResponsesAPIStreamingIterator.__next__cCs|j}|jdur%t|jdr%z t|j|j}Wn ty$Ynwt|jj||j t ddt j |jj|d|j t d|jt ddS)z6Handle logging for completed responses in sync contextNr)rcrEr$rarrr)r,r\r@rrrPr rrr$rr.rrkrrfrr6r6r7rQ s0  zDSyncResponsesAPIStreamingIterator._handle_logging_completed_responserl)rnrorprqrrrsrLrrtrrrr8rrrQrr6r6r|r7rs4  &rcseZdZdZdZ    ddejdedede de e ee fd e ed e e ee fd e effd d Z ddZdefddZddZdefddZdedefddZZS)!MockResponsesAPIStreamingIteratoru Mock iterator—fake a stream by slicing the full response text into 5 char deltas, then emit a completed event. Models like o1-pro don't support streaming, so we fake it. Nrrrrr r!r"r#c stj||||||||djjj||dfddtdtjD} t j r\|dur\t dd} | dur\z|j d} | durQt | d| Wn ty[Ynw| ttjd g_d_dS) N)rrrrr r!r"r#)r raw_responserc s0g|]}ttj||jjdddqS)r)r@deltaZitem_idZ output_indexZ content_index)rrZOUTPUT_TEXT_DELTA CHUNK_SIZEr().0i full_textr5Z transformedr6r7 nsz>MockResponsesAPIStreamingIterator.__init__..rrCrDrF)r@r)ryr8rZtransform_response_api_responser _collect_textrangelenrr;rNr-rOrKrPrrrM_events_idx) r5rrrrr r!r"r#ZdeltasrUrFr|rr7r8MsT       z*MockResponsesAPIStreamingIterator.__init__cCrrxr6rXr6r6r7rrz+MockResponsesAPIStreamingIterator.__aiter__r9cs4|jt|jkr t|j|j}|jd7_|SN)rrrrr5Zevtr6r6r7rs  z+MockResponsesAPIStreamingIterator.__anext__cCrrxr6rXr6r6r7rrz*MockResponsesAPIStreamingIterator.__iter__cCs2|jt|jkr t|j|j}|jd7_|Sr)rrrrrr6r6r7rs  z*MockResponsesAPIStreamingIterator.__next__respcCsDd}|jD]}t|dd}|dkrt|dgD]}||j7}qq|S)Nr%r@r:content)outputr-text)r5routZout_item item_typecr6r6r7rs   z/MockResponsesAPIStreamingIterator._collect_textrl)rnrorprqrrrrsrLrrtrrrr8rrrrrrrrr6r6r|r7rCs: Er)verbose_logger)zresponse.createdresponse.completedzresponse.failedzresponse.incompleteerrorc @seZdZdZ  ddedededeedeef dd Zd e d e fd d Z ded dfddZ ded dfddZ ded dfddZdddZdddZdddZdddZdS) ResponsesWebSocketStreaminga Manages bidirectional WebSocket forwarding for the Responses API WebSocket mode (wss://.../v1/responses). Unlike the Realtime API, the Responses API WebSocket mode: - Uses response.create as the client-to-server event - Streams back the same events as the HTTP streaming Responses API - Supports previous_response_id for incremental continuation - Supports generate: false for warmup - One response at a time per connection (sequential, no multiplexing) N websocket backend_wsruser_api_key_dictr"cCs2||_||_||_||_|pi|_g|_g|_dSrx)rrrrr"messagesinput_messages)r5rrrrr"r6r6r7r8s  z$ResponsesWebSocketStreaming.__init__ event_objr9cCs|dtvS)Nr@)r3RESPONSES_WS_LOGGED_EVENT_TYPES)r5rr6r6r7_should_store_eventsz/ResponsesWebSocketStreaming._should_store_eventeventc Csjt|tr |d}t|tr$zt|}Wntjtfy#YdSw|}||r3|j |dSdS)Nutf-8) rIbytesdecoderLrGrHrR TypeErrorrrappend)r5rrr6r6r7 _store_events    z(ResponsesWebSocketStreaming._store_eventr:c CsLzt|tr t|}n t|tr|}nWdS|ddkr!WdS|dg}t|tr8|jd|dWdSt|tr|D]U}t|tsGq?|ddkr|ddkr|d g}t|trj|jd|dq?t|tr|D]}t|tr|dd kr|d d }|r|jd|dqqq?WdSWdStj t t fyYdSw) zZlitellm_call_idr aresponsesZlitellm_logging_objZ_aresponses_websocket_MANAGED_WS_SKIP_KWARGSc@seZdZdZ      d@dededddeedeeeefd eed eed eed eed eddfddZ e dedeefddZ dAdededdfddZ dede eeeffddZdede eeefddfddZe d eeefdeefd!d"Ze d eeefde eeeffd#d$Ze d%ede eeeffd&d'Zd(edeeeeffd)d*Ze d+eeefdeeeffd,d-Zd.eeefdeed/e eeefd0e eeefddf d1d2Zd.eeefd3eeddfd4d5Ze d.eeefdeddfd6d7Zded.eeefdeeeeffd8d9Zd eeeefd0e eeefd/e eeefddfd:d;Zd(eddfdd?ZdS)C ManagedResponsesWebSocketHandlera Handles Responses API WebSocket mode for providers that do not expose a native ``wss://`` responses endpoint. Instead of proxying to a provider WebSocket, this handler: - Listens for ``response.create`` events from the client - Makes HTTP streaming calls via ``litellm.aresponses(stream=True)`` - Serialises and forwards every streaming event back over the WebSocket - Supports ``previous_response_id`` for multi-turn conversations via in-memory session tracking (avoids async DB-write timing issues) - Supports sequential requests over a single persistent connection This makes every provider that LiteLLM can reach over HTTP available on the WebSocket transport without any provider-specific changes. Nrrrrtrr rr*timeoutr!rbr9c KsX||_||_||_||_|pi|_||_||_||_| |_dd| D|_ i|_ dS)NcSsi|] \}}|tvr||qSr6)rrkvr6r6r7 sz=ManagedResponsesWebSocketHandler.__init__..) rrrrr rr*rr!items extra_kwargs_session_history) r5rrrrr rr*rr!rbr6r6r7r8s   z)ManagedResponsesWebSocketHandler.__init__rTc Csz2t|dr |jddWSt|drtj|jddtdWSt|tr+tj|tdWStt|WStyK}z t d|WYd}~dSd}~ww)zHSerialize a streaming chunk to a JSON string for WebSocket transmission.model_dump_jsonT)Z exclude_noner)defaultz1ManagedResponsesWS: failed to serialize chunk: %sN) r\rrGdumpsrrLrIrJrPrr)rTexcr6r6r7_serialize_chunks    z1ManagedResponsesWebSocketHandler._serialize_chunk server_errorr: error_typecsBz|jtd||ddIdHWdSty YdSw)Nr)r@r:)r@r)rrrGrrP)r5r:rr6r6r7 _send_errors z,ManagedResponsesWebSocketHandler._send_errorprevious_response_idcCs(t|}|d|}t|j|gS)z Return accumulated message history for *previous_response_id*. The key is the *decoded* response ID (the raw provider response ID before LiteLLM base64-encodes it into the ``resp_...`` format). response_id)r!_decode_responses_api_response_idr3rr)r5rdecodedZraw_idr6r6r7_get_history_messagess  z6ManagedResponsesWebSocketHandler._get_history_messagesrrcCs||j|<dS)u Store the complete accumulated message history for *response_id*. Replaces any prior value — callers are responsible for passing the full history (prior turns + current input + new output). N)r)r5rrr6r6r7_store_historysz/ManagedResponsesWebSocketHandler._store_historycompleted_eventcCsB|di}t|tr|dnd}|sdSt|}|d|S)z Pull the raw (decoded) response ID out of a ``response.completed`` event. Returns *None* if the event doesn't contain a usable ID. rr(Nr)r3rIrJrr)rresp_objZ encoded_idrr6r6r7_extract_response_ids   z5ManagedResponsesWebSocketHandler._extract_response_idc Cs|di}t|ts gSg}|dgpgD]C}t|tsq|d}|dd}|dkrQ|dp4g}dd |D}d |}|rP|d|d |d gd q|dkrZ||q|S)z Convert the output items in a ``response.completed`` event into Responses API message dicts suitable for the next turn's ``input``. rrr@rZ assistantr:rcSs0g|]}t|tr|ddvr|ddqS)r@) output_textrrr%)rIrJr3)rpr6r6r7rs zMManagedResponsesWebSocketHandler._extract_output_messages..r%rr@rr@rrZ function_call)r3rIrJjoinr) rrrrArrZ content_partsZ text_partsrr6r6r7_extract_output_messagess,       z9ManagedResponsesWebSocketHandler._extract_output_messages input_valcCs<t|trddd|dgdgSt|trdd|DSgS)z Normalise the ``input`` field of a ``response.create`` event to a list of Responses API message dicts. r:rrrrcSsg|] }t|tr|qSr6)rIrJ)rrAr6r6r7rszGManagedResponsesWebSocketHandler._input_to_messages..)rIrLr)rr6r6r7_input_to_messagess  z3ManagedResponsesWebSocketHandler._input_to_messages raw_messagecsPzt|}Wntjy|ddIdHYdSw|ddkr&dS|S)zOParse raw WS text; return the message dict or None (JSON error / ignored type).z%Invalid JSON in response.create eventZinvalid_request_errorNr@r)rGrHrRrr3)r5rrr6r6r7_parse_messagesz/ManagedResponsesWebSocketHandler._parse_messagercs@|d}t|tr|r|ndd|DfddtDS)z Extract Responses API params from the event, handling both wire formats: Nested: {"type": "response.create", "response": {"input": [...], ...}} Flat: {"type": "response.create", "input": [...], "model": "...", ...} rcSsi|] \}}|dkr||qS)r@r6rr6r6r7r-szLManagedResponsesWebSocketHandler._build_base_call_kwargs..cs*i|]}|vr|dur||qSrxr6)rparamZresponse_paramsr6r7r/s )r3rIrJrr)rnestedr6rr7_build_base_call_kwargs"s  z8ManagedResponsesWebSocketHandler._build_base_call_kwargs call_kwargscurrent_messages prior_historycCsF|sdS|r|||d<tdt||dStd|||d<dS)zHPrepend in-memory turn history, or fall back to DB-based reconstruction.NrzMManagedResponsesWS: prepended %d history messages for previous_response_id=%szuManagedResponsesWS: no in-memory history for previous_response_id=%s; falling back to DB-based session reconstructionr)rrr)r5r rr r r6r6r7_apply_history5s  z/ManagedResponsesWebSocketHandler._apply_history event_modelcCsp|jdur |j|d<|jdur|j|d<|jdur|j|d<|jdur*|s*|j|d<|jr6t|j|d<dSdS)zBInject connection-level credentials and metadata into call_kwargs.Nrr*rr!r )rr*rr!r rJ)r5r rr6r6r7_inject_credentialsPs       z4ManagedResponsesWebSocketHandler._inject_credentialscCs|dpidp i}t|tsdSt|dpi}|d|d<|d|d<||d<dD]}||vrB||durB||||<q0i|d|i}d|vrSi|d<||dd<|d i||d d<dS) zGUpdate proxy_server_request body so spend logs record the full request.r proxy_server_requestNbodyrstorer)ZtoolsZ tool_choiceZ instructionsmetadatar&)r3rIrJ setdefault)r rrrrr6r6r7_update_proxy_requestbs*    z6ManagedResponsesWebSocketHandler._update_proxy_requestc sd}tjdd|i|IdH}|2zg3dHW}|durqt|ddp.t|tr-|dnd}||}|dur9q|dkrR|durRzt|}Wn t yQYnwz |j |IdHWqt yy}zt d||WYd}~Sd}~ww6|S)a> Stream ``litellm.aresponses`` and forward every chunk over the WebSocket. Captures the ``response.completed`` event type from the chunk object directly (before serialization) to avoid a redundant JSON round-trip on every chunk. Returns the completed event dict, or ``None``. Nrr@rz5ManagedResponsesWS: error sending chunk to client: %sr6)r;rr-rIrJr3rrGrHrPrrrr) r5rr rZstream_responserTZ chunk_typeZ serializedZsend_excr6r6r7_stream_and_forwardxs:    z4ManagedResponsesWebSocketHandler._stream_and_forwardcCsV|durdS||}|sdS||}|||}|||tdt||dS)zMStore this turn in in-memory history for future previous_response_id lookups.Nz9ManagedResponsesWS: stored %d messages for response_id=%s)rrrrrr)r5rr r Znew_response_idZ output_msgsZ all_messagesr6r6r7_save_turn_historys    z3ManagedResponsesWebSocketHandler._save_turn_historyc s||IdH}|durdS||}d|d<|dd}|p"|j}|dd}||d}|r8||ng}||||||||| ||| |j z | ||IdH} Wn#t y} ztd| |t| IdHWYd} ~ dSd} ~ ww|| ||dS)a Parse one ``response.create`` event, call ``litellm.aresponses(stream=True)``, and forward every streaming event to the client. Multi-turn support via in-memory session history ------------------------------------------------ When ``previous_response_id`` is present in the event: 1. Look up the accumulated message history in ``self._session_history`` (keyed by the decoded provider response ID). 2. Prepend those messages to the current ``input`` so the model has full conversation context. 3. After the stream completes, extract the new response ID and output messages from ``response.completed`` and store them in ``self._session_history`` for the next turn. This in-memory approach avoids the async DB-write race condition that occurs when spend logs haven't been committed by the time the second ``response.create`` arrives over the same WebSocket connection. NTstreamrrrz8ManagedResponsesWS: error processing response.create: %s)rr poprrr3rr rrrdrrrPrrgrrLr) r5rrr rrrr r rrr6r6r7_process_response_creates:        z9ManagedResponsesWebSocketHandler._process_response_createc sz/ z |jIdH}Wnty'}ztd|WYd}~WdSd}~ww||IdHqtyT}ztd||d|IdHWYd}~dSd}~ww)z Main loop: accept ``response.create`` events sequentially and handle each one before waiting for the next message. TNz+ManagedResponsesWS: client disconnected: %sz(ManagedResponsesWS: unexpected error: %szInternal server error: )rrrPrrrrgr)r5r:rr6r6r7runs& $z$ManagedResponsesWebSocketHandler.run)NNNNNN)rrm)rnrorprqrrLrrfloatr8 staticmethodrrrrrrrrrr r rrrrrrr6r6r6r7rzs    $"   (  $       $ rrGr0rirtypingrrrrrrr;Zlitellm.constantsrrZ#litellm.litellm_core_utils.asyncifyr Z'litellm.litellm_core_utils.core_helpersr Z*litellm.litellm_core_utils.litellm_loggingr rtZ:litellm.litellm_core_utils.llm_response_utils.get_api_baser Z?litellm.litellm_core_utils.llm_response_utils.response_metadatar Z/litellm.litellm_core_utils.thread_pool_executorrZ.litellm.llms.base_llm.responses.transformationrZlitellm.responses.utilsrZlitellm.types.llms.openairrrrrrrZlitellm.types.utilsrZ litellm.utilsrrrr`rvrrZlitellm._loggingrrrr__required_keys____optional_keys__r frozenset__annotations__rrr6r6r6r7sL          $ D ih q   -