
    0j0                         d Z ddlZddlZddlmZ ddlmZmZmZm	Z	 ddl
mZ ddlmZmZmZmZmZmZ ddlmZ dd	lmZmZmZ dd
lmZmZ  ej        e          Z G d de          ZdS )z[
Copyright (c) Microsoft Corporation. All rights reserved.
Licensed under the MIT License.
    N)deque)	AwaitableCallableOptionalUnion)HTTPStatusError)	ApiClientChannelDataConversationReferenceMessageActivityInputSentActivityTypingActivityInput)EventEmitter   )StreamCancelledErrorStreamerEventStreamerProtocol)RetryOptionsretryc                       e Zd ZdZdedef fdZddZede	fd            Z
ede	fd	            Zedefd
            Zedefd            Zdeeged         f         fdZdeeged         f         fdZdeeeef         ddfdZdeddfdZddZd Zdee         fdZddZdefdZdeeef         defdZ  xZ!S )
HttpStreama  
    HTTP-based streaming implementation for Microsoft Teams activities.

    Flow:
    1. emit() adds activities to a queue
    2. _flush() drains the entire queue under a lock.
    3. Informative typing updates are sent immediately if no message started.
    4. Message text are combined into a typing chunk.
    5. Another flush is scheduled if more items remain.
    6. close() waits for queue to empty, then sends final message with stream_type='stream_final'

    The timeout cancellation ensures only one flush operation is scheduled at a time.
    The delays between flushes is to ensure we dont hit API rate limits with Microsoft Teams.
    clientrefc                 n   t                                                       || _        || _        t	          t
                               | _        d| _        t          j	                    | _
        d| _        d| _        d| _        t          j                    | _        d| _        |                                  dS )z
        Initialize a new HttpStream instance.

        Args:
            client (ApiClient): The API client used to send activities to Microsoft Teams.
            ref (ConversationReference): Reference to the Teams conversation.
        Ng      >@F)super__init___client_refr   r   _events_resultasyncioLock_lock_timeout_pending_total_wait_timeoutEvent_state_changed	_canceled_reset_state)selfr   r   	__class__s      h/home/rurouni/.hermes/hermes-agent/venv/lib/python3.11/site-packages/microsoft_teams/apps/http_stream.pyr   zHttpStream.__init__,   s     		#M244/3\^^
7;6:*. %moo    returnNc                     d| _         d| _        d| _        t                      | _        d| _        t                      | _        dS )z)Reset the stream state to initial values.r   N )_index_id_textr
   _channel_data_final_activityr   _queuer+   s    r-   r*   zHttpStream._reset_stateC   s<    "&
*5--?CTYT[T[r.   c                     | j         S )z
        Whether the stream has been canceled.
        For example when the user pressed the Stop button or the 2-minute timeout has exceeded.
        )r)   r8   s    r-   canceledzHttpStream.canceledL   s     ~r.   c                     | j         duS )z/Whether the final stream message has been sent.N)r    r8   s    r-   closedzHttpStream.closedT   s     |4''r.   c                 *    t          | j                  S )z-The total number of chunks queued to be sent.)lenr7   r8   s    r-   countzHttpStream.countY   s     4;r.   c                     | j         S )zGThe sequence number, representing the number of stream activities sent.)r2   r8   s    r-   sequencezHttpStream.sequence^   s     {r.   handlerc                 <    | j                             d|           d S )Nchunk)r   onr+   rB   s     r-   on_chunkzHttpStream.on_chunkc   s    )))))r.   c                 <    | j                             d|           d S )Nclose)r   oncerF   s     r-   on_closezHttpStream.on_closef   s     '7+++++r.   activityc                 ,   | j         rt          d          t          |t                    rt	          |d          }| j                            |           | j        s4| j        s/t          j
        |                                           | _        dS dS dS )zo
        Emit a new activity to the stream.

        Args:
            activity: The activity to emit.
        zStream has been cancelled.message)texttypeN)r)   r   
isinstancestrr   r7   appendr%   r$   r!   create_task_flush)r+   rL   s     r-   emitzHttpStream.emiti   s     > 	E&'CDDDh$$ 	K+	JJJH8$$$} 	?T] 	?#/>>DMMM	? 	? 	? 	?r.   rO   c                     |                      t                                          |                              t	          d                               dS )z
        Send status updates before emitting (ex. "Thinking...").

        Args:
            text: The status text to send.
        informative)stream_typeN)rV   r   	with_textwith_channel_datar
   )r+   rO   s     r-   updatezHttpStream.update|   sK     			%''11$77II+boJpJpJpqqrrrrrr.   c                 h    d| _         t          d | j        D                       | _        d| _        dS )a  
        Discard everything accumulated so far so the caller can emit a fresh
        card-only final activity.

        Clears the text buffer, drops queued message items so the next flush
        doesn't repopulate it, and discards the promoted final activity so any
        previously-flushed attachments/suggested actions aren't sent if the
        caller never emits a replacement. The stream id and channel data are
        kept intact so the new final activity still updates the stream in place.
        r1   c              3   D   K   | ]}t          |t                    |V  d S N)rQ   r   ).0items     r-   	<genexpr>z(HttpStream.clear_text.<locals>.<genexpr>   s2      ggT
4Qe@f@fgDggggggr.   N)r4   r   r7   r6   r8   s    r-   
clear_textzHttpStream.clear_text   s9     
ggT[ggggg#r.   c                     K    fd}	 t          j         |             j                   d{V  dS # t           j        $ r Y dS w xY w)z]Wait until _id is set, the queue is empty, and no flush is in progress, with a total timeout.c                  @  K    j         s  j        r j                                        rq j        sh j                                         d {V   j                                          j         s  j        r j                                        r j        fd S d S d S d S r_   )r7   r3   r#   lockedr)   r(   waitclearr8   s   r-   _pollz0HttpStream._wait_for_id_and_queue.<locals>._poll   s      ; ,dh ,$*2C2C2E2E ,t~ ,)..000000000#))+++ ; ,dh ,$*2C2C2E2E ,t~ , , , , , , , , ,r.   )timeoutNTF)r!   wait_forr&   TimeoutError)r+   ri   s   ` r-   _wait_for_id_and_queuez!HttpStream._wait_for_id_and_queue   s      	, 	, 	, 	, 	,
	"5577D4LMMMMMMMMMM4# 	 	 	55	s   )5 AAc                 4   K    j         !t                              d            j         S  j        rt                              d           d S  j        dk    r< j        s5 j                                        st                              d           d S                                   d {V }|st          	                    d           d S  j
        dk    p% j        o j        j        p j        o j        j        }|st          	                    d           d S  j        
J d             j        pt                                           j
                                       j                                       j                                                   t+           fd	t-                      
           d {V } j                            d|                                             | _         t                              d|           |S )Nz!stream already closed with resultz&stream was cancelled, nothing to closer   z-stream has no content to send, returning NonezfTimeout while waiting for _id to be set, queue to be empty, and flush to complete, cannot close streamr1   zGno text, attachments, or suggested actions to send, cannot close streamzID should be set by this pointc                  .                                    S r_   _send)rL   r+   s   r-   <lambda>z"HttpStream.close.<locals>.<lambda>   s    $**X"6"6 r.   optionsrI   zstream closed with result: %s)r    loggerdebugr)   r2   r7   r#   rf   rm   warningr4   r6   attachmentssuggested_actionsr3   r   rZ   with_idr[   r5   add_stream_finalr   r   r   rV   r*   )r+   resulthas_contentresrL   s   `   @r-   rI   zHttpStream.close   s.     <#LL<===<> 	LLABBB4;!DK
8I8I8K8KLLHIII4 2244444444 	NNx   4 J" Q$I)=)IQ$O)=)O 	
  	NNdeee4 x##%E###'A+?+A+A4:&&..tx88JJ4K]^^ooqqq66666OOOOOOOOO 	'3''' 	4c:::
r.   c                 v   K    j                                         rdS  j                                          d{V  	  j        s<	 d _         j                                           j                                         dS  j          j        	                                 d _        g }t           j                  } j        r j                                        }t          |t                    r xj        |j        pdz  c_        | _        t          |t          t"          f          rL|j        rEi  j                                        |j                                        }t+          d
i | _        t          |t"                    r:t-          |j        dd          dk    r  j        dk    r|                    |            j        |dk    rVt0                              d           	 d _         j                                           j                                         dS |D ]}                     |           d{V   j        r0t#           j                  }                     |           d{V   j        r6 j        s/t7          j                                        d fd	           _        d _         j                                           j                                         dS # d _         j                                           j                                         w xY w)z3
        Flush the current activity queue.
        Nr1   rY   rX   r   zNo activities to flush)rO   g      ?c                  P    t          j                                                   S r_   )r!   rT   rU   r8   s   r-   rr   z#HttpStream._flush.<locals>.<lambda>
  s     SZSfgkgrgrgtgtSuSu r.    )r#   rf   acquirer7   r%   releaser(   setr$   cancelr>   popleftrQ   r   r4   rO   r6   r   channel_datar5   
model_dumpr
   getattrrS   ru   rv   _send_activityr!   get_running_loop
call_later)r+   informative_updatesstart_lengthrL   mergedtyping_updateto_sends   `      r-   rU   zHttpStream._flush   sm     
 : 	Fj  """""""""1	&; Z !DMJ   ##%%%%%] }($$&&& $=?t{++L+ 9;..00h(<== 4JJ(-"525JJ+3D(h)=?R(STT ?YaYn ?f 2 = = ? ?f8CXCcCcCeCefF)4)>)>v)>)>D&x)<==9 5}dKK}\\
b(( (..x888! + 9$ q  5666" !DMJ   ##%%%%%! "5 9 9))-8888888888 z 3-4:>>>))'222222222 { w4= w ' 8 : : E EcKuKuKuKu v v !DMJ   ##%%%%% !DMJ   ##%%%%s   K= E0K= .BK= =;L8r   c                    K    j         r                     j                                        j                  t	           fdt          ddd                     d{V } j                            d|            xj        d	z  c_         j         '|j         _          j	        
                                 dS dS )
z
        Send an activity to the Teams conversation with the ID.

        Args:
            activity: The activity to send.
        c                  .                                    S r_   rp   )r+   r   s   r-   rr   z+HttpStream._send_activity.<locals>.<lambda>  s    DJJw'' r.   g      @none   )	max_delayjitter_typemax_attemptsrs   NrD   r   )r3   rz   add_stream_updater2   r   r   r   rV   idr(   r   )r+   r   r~   s   `` r-   r   zHttpStream._send_activity  s       8 	0oodh//G++DK88''''' 3FQRSSS
 
 
 
 
 
 
 
 
 	'3'''q8vDH##%%%%% r.   c                   K   | j         r)t                              d           t          d          | j        j        |_        | j        j        |_        	 |j        rnt          d |j
        pg D                       sN| j        j                            | j        j        j                                      |j        |           d{V }nG| j        j                            | j        j        j                                      |           d{V }t!          j        ||          S # t$          $ rG}|j        j        dk    r1d| _         t                              d           t          d          | d}~ww xY w)z
        Send or update an activity to the Teams conversation.

        Args:
            activity: The activity to send.
        z!Teams channel stopped the stream.c              3   ,   K   | ]}|j         d k    V  dS )
streaminfoN)rP   )r`   es     r-   rb   z#HttpStream._send.<locals>.<genexpr>6  s)      %_%_af&<%_%_%_%_%_%_r.   Ni  T)r)   ru   rw   r   r   botfrom_conversationr   anyentitiesr   conversations
activitiesr\   creater   merger   responsestatus_code)r+   r   r~   r   s       r-   rq   zHttpStream._send'  s      > 	LNN>???&'JKKK	#y5	z m#%_%_wGWG][]%_%_%_"_"_ m L6AA$)BXB[\\ccdkdnpwxxxxxxxx L6AA$)BXB[\\ccdkllllllll%gs333 	 	 	z%,,!%BCCC*+NOOUVV	s   CD' '
E81AE33E8)r/   N)"__name__
__module____qualname____doc__r	   r   r   r*   propertyboolr:   r<   intr?   rA   r   r   r   rG   rK   r   r   r   rR   rV   r\   rc   rm   r   rI   rU   r   rq   __classcell__)r,   s   @r-   r   r      s]        y /D      .\ \ \ \ $    X ( ( ( ( X(  s       X  #    X*,4)H I * * * *,,4)H I , , , ,?U#79Lc#QR ?W[ ? ? ? ?&s3 s4 s s s s$ $ $ $$  .Xl3 . . . .`;& ;& ;& ;&z&,? & & & &*5)<>R)R#S Xd        r.   r   ) r   r!   loggingcollectionsr   typingr   r   r   r   httpxr   microsoft_teams.apir	   r
   r   r   r   r   microsoft_teams.commonr   plugins.streamerr   r   r   utilsr   r   	getLoggerr   ru   r   r   r.   r-   <module>r      sV   
         7 7 7 7 7 7 7 7 7 7 7 7 ! ! ! ! ! !                0 / / / / / S S S S S S S S S S & & & & & & & &		8	$	$e e e e e! e e e e er.   