idZddlmZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z ej dZeZeZeZe GddZGd d ZdS) u~Gateway streaming consumer — bridges sync agent callbacks to async platform delivery. The agent fires stream_delta_callback(text) synchronously from its worker thread. GatewayStreamConsumer: 1. Receives deltas via on_delta() (thread-safe, sync) 2. Queues them to an asyncio task via queue.Queue 3. The async run() task buffers, rate-limits, and progressively edits a single message on the target platform Design: Uses the edit transport (send initial message, then editMessageText). This is universally supported across Telegram, Discord, and Slack. Credit: jobless0x (#774, #1312), OutThisLife (#798), clicksingh (#697). ) annotationsN) dataclass)AnyOptionalzgateway.stream_consumercZeZdZUdZdZded<dZded<dZd ed <d Zd ed <dZ ded<dS)StreamConsumerConfigz5Runtime config for a single stream consumer instance.g?float edit_interval(intbuffer_thresholdu ▉strcursorFbool buffer_onlyfresh_final_after_secondsN) __name__ __module__ __qualname____doc__r __annotations__r rrrdZd=dZdzz z z z )zzz z z z Nadapterrchat_idrconfigOptional[StreamConsumerConfig]metadataOptional[dict]on_new_messageOptional[callable]c||_||_|p t|_||_||_t j|_d|_ d|_ d|_ d|_ d|_ d|_d|_d|_d|_d|_|jj|_d|_t-|dddu|_d|_d|_dS)NFTrrREQUIRES_EDIT_FINALIZE)rr rcfgr#_on_new_messagequeueQueue_queue _accumulated _message_id_message_created_ts _already_sent_edit_supported_last_edit_time_last_sent_text_fallback_final_send_fallback_prefix_flood_strikesr _current_edit_interval_final_response_sentgetattr_adapter_requires_finalize_in_think_block _think_buffer)selfrr r!r#r%s r__init__zGatewayStreamConsumer.__init__Xs  3133   .#(;== *. 59 "#"!$)! "&*h&<#$)! G5u = = E ' %rreturnrc|jS)z?True if at least one message was sent or edited during the run.)r2r?s r already_sentz"GatewayStreamConsumer.already_sents !!rc|jS)zBTrue when the stream consumer delivered the final assistant reply.)r:rCs rfinal_response_sentz)GatewayStreamConsumer.final_response_sents ((rNonecD|jtdS)z>Finalize the current stream segment and start a fresh message.N)r.put _NEW_SEGMENTrCs ron_segment_breakz&GatewayStreamConsumer.on_segment_breaks  %%%%%rtextcP|r#|jt|fdSdS)z7Queue a completed interim assistant commentary message.N)r.rI _COMMENTARYr?rLs r on_commentaryz#GatewayStreamConsumer.on_commentarys3  1 KOO[$/ 0 0 0 0 0 1 1rc|j}|dS |dS#t$r tddYdSwxYw)z8Fire the on_new_message callback, swallowing any errors.Nzon_new_message callback errorT)exc_info)r+ Exceptionloggerdebug)r?cbs r_notify_new_messagez)GatewayStreamConsumer._notify_new_messagesh  ! : F I BDDDDD I I I LL84L H H H H H H Is &AAFpreserve_no_editrYcx|r |jdkrdSd|_d|_d|_d|_d|_d|_dS)N __no_edit__r(F)r0r1r/r5r6r7)r?rYs r_reset_segment_statez*GatewayStreamConsumer._reset_segment_statesR   0M A A F#' !$)! "rcr|r|j|dS||dSdS)u2Thread-safe callback — called from the agent's worker thread. When *text* is ``None``, signals a tool boundary: the current message is finalized and subsequent text will be sent as a new message so it appears below any tool-progress messages the gateway sent in between. N)r.rIrKrOs ron_deltazGatewayStreamConsumer.on_deltasJ  $ KOOD ! ! ! ! ! \  ! ! # # # # #\rcD|jtdS)z#Signal that the stream is complete.N)r.rI_DONErCs rfinishzGatewayStreamConsumer.finishs rc|j|z}d|_|rm|jrd}d}|jD]:}||}|dkr|dks||kr|}t |};|rd|_|||zd}nt d|jD}t ||kr || dn||_dSd}d}|jD]}d} |||}|dkrn|dkr#|j p|jd} n|d|} | d} | dkr;|j s|jdo| dk} n#| | d zd dk} | r|dks||kr|}t |}n|d z}|r-|xj|d|z c_d|_|||zd}nd} |jD]G}td t |D]'} ||d| r| | kr| } (H| r*|xj|d| z c_|| d|_n|xj|z c_dS|kdSdS) abAdd a text delta to the accumulated buffer, suppressing think blocks. Uses a state machine that tracks whether we are inside a reasoning/thinking block. Text inside such blocks is silently discarded. Partial tags at buffer boundaries are held back in ``_think_buffer`` until enough characters arrive to decide. r(rFNc34K|]}t|VdS)N)len).0ts r z?GatewayStreamConsumer._filter_and_accumulate..s(!I!IQ#a&&!I!I!I!I!I!IrT ) r>r=_CLOSE_THINK_TAGSfindremax_OPEN_THINK_TAGSr/endswithrfindstriprange)r?rLbufbest_idxbest_lentagidxmax_tag search_start is_boundary precedinglast_nl held_backis r_filter_and_accumulatez,GatewayStreamConsumer._filter_and_accumulatesd 4'K #J 1,,C((3--Cbyyh"nnh#&#&s88 +0D(h1223CC"!I!I$2H!I!I!IIIG;>s88g;M;MgXYYSVD&F 0//C#$L/!hhsL99"99!!88$($5 5!D#'#4#=#=d#C#C(K ),DSD I&/ood&;&;G&"}})-):%:&I(,(9(B(B4(H(H%@(1(9(9R(?!, /8! .E.K.K.M.MQS.S &"HNNcHnn'*H'*3xxH!'*Qw 5/8%%YhY7%%+/D(h1223CC!"I#4..!&q#c((!3!3..A"||CG44.Y,- .!1))S9*-==))-0)-=**))S0))FWK K K K K rcb|jr%|js |xj|jz c_d|_dSdSdS)zFlush any held-back partial-tag buffer into accumulated text. Called when the stream ends (got_done) so that partial text that was held back waiting for a possible opening tag is not lost. r(N)r>r=r/rCs r_flush_think_bufferz)GatewayStreamConsumer._flush_think_buffer"sR   $d&: $   !3 3  !#D    $ $ $ $rc Kt|jdd}td|t|jjz dz } d}d}d} |j}|turd}nx|turd}nlt|tr+t|dkr|d tur |d }n,| |n#tj$rYnwxYw|r|t#j}||jz }|p|p|du} |jjs6| p3||jkr|jp!t|j|jjk} d} | r|jrt|j|kr|j|j|j|} | D]#} || |jd{V$d |_d |_t#j|_|r|j|_dS|rd|_d|_d |_t|j|kr|j|j r|j!d d |} | |dzkr|} |jd| } |"| d{V}|js|sn[|j| d#d |_d|_d |_t|j|kr|j|j |j}|s|s|||jjz }|"|| d{V} t#j|_|r|jr|jr!|$|jd{Vnl| r|j%sd|_n[|jr(|"|jd d{V|_n,|js%|"|jd{V|_dS|[|&|'|d{Vt#j|_|&|rK|jr.| s,|jr%|jdkr|(d{V|&dtSj*dd{Vu#tRj+$rid}|jrF|jr? tY|"|jd{V}n#tZ$rYnwxYw|r|js d|_YdSYdSYdStZ$r&}t\/d|Yd}~dSd}~wwxYw)z@Async task that drains the queue and edits the platform message.MAX_MESSAGE_LENGTHdTFNrrjr(rifinalizer[rXg?zStream consumer error: %s)0r;rrmrer*rr. get_nowaitr`rJ isinstancetuplerNrr,Emptyrtime monotonicr4rr9r/r r0truncate_message_send_new_chunkr5r2r:r6r7r3rp _send_or_editlstrip_send_fallback_finalr<r\_send_commentary#_flush_segment_tail_on_edit_failureasynciosleepCancelledErrorrrSrTerror)r? _raw_limit _safe_limitgot_donegot_segment_breakcommentary_textitemnowelapsed should_editcurrent_update_visiblechunkschunksplit_atok display_text_best_effort_okes rrunzGatewayStreamConsumer.run,s\T\+?FF #zC,@,@@3FGG F 9q * $)!"& #{55775=='+H!<//04-!%dE22"s4yyA~~$q'U`J`J`.21gO!33D9999 ;&/,,...n&& 443(3&d2 x+"-# D$??2 $ 1Ot011TX5NN */&B<4#4B<D-..<< ,4"&!>!> -{""&,PPE"&"6"6ud>N"O"OOOOOOOOO,.)/1,/3~/?/?,##8<8JD5"F,7/3D,8=D546D1 D-..<< ,8 09$(#4#:#:4K#P#P#kQ&666'2H $ 1)8) <#'#5#5e#<#<<<<<<<4"B" ",0,=hii,H,O,OPT,U,U)+/(/1,%D-..<< ,8 09&$(#4L#8,=8/BY$7 483E3E$!24F44......*,0>+;+;D( (d4d"&";";DP>P $ 1D?Q??999999D55"&!3d>B>P>PQUQb>c>c8c8c8c8c8c8cD5F".--/////@@@@@@@@@+/>+;+;D(--/// %E)I 6I!,I!, =="FFHHHHHHHHH--t-DDDmD)))))))))cq *f% 1 1 1#O  T%5 &*1C1CDDU1V1V+V+V+V+V+V+V&W&WOO D 1t'@ 1,0)))) 1 1 1 1 1 1 9 9 9 LL4a 8 8 8 8 8 8 8 8 8 9sR=$C4R=5 CR=?CR=CR=C)&R=(C))D'R=G#R=7CR==U#-T  U# TU#TU#4 U#=UU#z[`"']?MEDIA:\s*\S+[`"']?cd|vrd|vr|S|dd}tjd|}t jdd|}|S)uStrip MEDIA: directives and internal markers from text before display. The streaming path delivers raw text chunks that may include ``MEDIA:`` tags and ``[[audio_as_voice]]`` directives meant for the platform adapter's post-processing. The actual media files are delivered separately via ``_deliver_media_from_response()`` after the stream finishes — we just need to hide the raw directives from the user. zMEDIA:z[[audio_as_voice]]r(z\n{3,}z )replacer _MEDIA_REsubrerstrip)rLcleaneds r_clean_for_displayz(GatewayStreamConsumer._clean_for_displaysn 4  $8$D$DK,,3R88'155b'BB&FG44~~r reply_to_id Optional[str]cK||}|s|S |jrt|jni}|j|j|||d{V}|jrV|jrOt|j|_ d|_ ||_ | t|jSd|_|S#t$r'}t d||cYd}~Sd}~wwxYw)zSend a new message chunk, optionally threaded to a previous message. Returns the message_id so callers can thread subsequent chunks. )r contentreply_tor#NTFzStream send chunk error: %s)rrqr#dictrsendr success message_idrr0r2r5rWr3rSrTr)r?rLrmetaresultrs rrz%GatewayStreamConsumer._send_new_chunksN &&t,,zz||   *.-?4 &&&RD<,, $ -F ~ #&"3 ##&v'8#9#9 %)"'+$((***6,---',$""    LL6 : : :       s$B"CC D %DD D c|jpd}|jjrA||jjr"|dt |jj }||S)z>Return the visible text already shown in the streamed message.r(N)r5r*rrorerr?prefixs r_visible_prefixz%GatewayStreamConsumer._visible_prefix2se%+ 8? 4vtx?? 42c$(/22223F&&v...r final_textc|jp|}|r>||r)|t|dS|S)zAReturn only the part of final_text the user has not already seen.N)r7r startswithrer)r?rrs r_continuation_textz(GatewayStreamConsumer._continuation_text9s_&@$*>*>*@*@  5j++F33 5c&kkll+2244 4rlimitr list[str]clt||kr|gSg}|}t||kro|dd|}||dzkr|}||d|||dd}t||ko|r|||S)z;Split text into reasonably sized chunks for fallback sends.rirrN)rerpappendr)rLrr remainingrs r_split_text_chunksz(GatewayStreamConsumer._split_text_chunks@s t99  6M )nnu$$ tQ66H%1*$$  MM)IXI. / / /!()),33D99I )nnu$$  % MM) $ $ $ rcvK||}||}d|_|s|r||kr|}n|jr|jr|jjr|j |jjrt|jdt|jj } |j |j |j|d{V}|jr||_n#t$rYnwxYwd|_d|_dSt%|j dd}t'd|dz }|||}d} d } d} |D]} d}t+d D]} |j |j | |j d{V}|jrnQ| d krJ||r5t2d t7jdd{V|r|jsG| r&d|_d|_| |_| |_d |_dSd|_d|_d |_d |_dSd} | } |jp| } || |_d|_d|_|d|_d |_dS)zSend the final continuation after streaming edits stop working. Retries each chunk once on flood-control failures with a short delay. FNr rrTrrrrr(rr rr#rz.Flood control on fallback send, retrying in 3sg@rc) rrr6rqrr0r5r*rrorer edit_messager rrSr2r:r;rmrrrrr#_is_flood_errorrTrUrrr7rrW)r?rLr continuation clean_textr raw_limit safe_limitrlast_message_idlast_successful_chunksent_any_chunkrattempts rrz*GatewayStreamConsumer._send_fallback_finalQs} ,,T22 ..z:: $)!!!##$ !! jD4H4H4J4J&J&J) $,,55dhoFF  "&!56LDHO8L8L7L6L!MJ '+|'@'@$(L'+'7$.(A(("""""" ">>3=D0$%)",0)DL*>EE i#o.. ((zBB)- "* '* 'EF 88  #|00 L!!] 1   >Ea<||r)|t |d}||}| sdS |j |j ||j d{V}|jr d|_dSdS#t $r&}t"d|Yd}~dSd}~wwxYw)aKDeliver un-sent tail content before a segment-break reset. When an edit fails (flood control, transport error) and a tool boundary arrives before the next retry, ``_accumulated`` holds text that was generated but never shown to the user. Without this flush, the segment reset would discard that tail and leave a frozen cursor in the partial message. Sends the tail that sits after the last successfully-delivered prefix as a new message, and best-effort strips the stuck cursor from the previous partial message. NrTz"Segment-break tail flush error: %s)r6_try_strip_cursorr7rr/rrerrrqrrr r#rr2rSrTr)r?visibletailrrs rrz9GatewayStreamConsumer._flush_segment_tail_on_edit_failuresv( +((** * * * * * * *'A4+?+?+A+A   0tw// 0G &--//D&&t,,zz||  F B<,, -F ~ *%)""" * * B B B LL=q A A A A A A A A A Bs2;C11 D!;DD!cK|jr |jdkrdS|}|r|sdS |j|j|j|d{V||_dS#t$rYdSwxYw)uBest-effort edit to remove the cursor from the last visible message. Called when entering fallback mode so the user doesn't see a stuck cursor (▉) in the partial message. r[Nr)r0rrqrrr r5rSrs rrz'GatewayStreamConsumer._try_strip_cursors  4#3}#D#D F%%'' V\\^^  F ,++ +,        $*D     DD s4A:: BBcbK||}|sdS |j|j||jd{V}|jr||jS#t$r&}t d|Yd}~dSd}~wwxYw)z6Send a completed interim assistant commentary message.FrNzCommentary send error: %s) rrqrrr r#rrWrSrTr)r?rLrrs rrz&GatewayStreamConsumer._send_commentarys&&t,,zz|| 5 <,, -F~ +((***> !    LL4a 8 8 855555 sAA>> B.B))B.ct|jddpd}|dkrdS|jr |jdkrdS|jdSt j|jz }||kS)aReturn True when a long-lived preview should be replaced with a fresh final message instead of an edit. Conditions: - Fresh-final is enabled (``fresh_final_after_seconds > 0``). - We have a real preview message id (not the ``__no_edit__`` sentinel and not ``None``). - The preview has been visible for at least the configured threshold. Ported from openclaw/openclaw#72038. rrrFr[)r;r*r0r1rr)r? thresholdages r_should_send_fresh_finalz.GatewayStreamConsumer._should_send_fresh_finalsvDH&A3GGN3 >>5 4#3}#D#D5  # +5n!99ircK|j} |j|j||jd{V}n3#t $r&}t d|Yd}~dSd}~wwxYwt|ddsdS|rj|dkrdt|jdd}|L ||j|d{Vn3#t $r&}t d||Yd}~nd}~wwxYwt|d d}|r ||_tj |_ nd|_d|_ d |_ ||_ d |_d S) aCSend ``text`` as a brand-new message (best-effort delete the old preview) so the platform's visible timestamp reflects completion time. Returns True on successful delivery, False on any failure so the caller falls back to the normal edit path. Ported from openclaw/openclaw#72038. rNz1Fresh-final send failed, falling back to edit: %sFrr[delete_messagez+Fresh-final preview cleanup failed (%s): %srT)r0rrr r#rSrTrUr;rrr1r2r5r:)r?rLold_message_idrr delete_fnnew_message_ids r_try_fresh_finalz&GatewayStreamConsumer._try_fresh_final$s) <,, -FF     LLLa P P P55555 vy%00 5  n == .>EEI$#)DL.AAAAAAAAAA LLE&!t<<  ,-D '+~'7'7D $ $ -D '+D $!#$(!ts--9 A)A$$A) B88 C(C##C(rrcK||}|}|jjr ||jjd}|}|sdS|sdSd}|j/|jjr#|jj|vrt ||krdS |j|jr||jkr |r|j sdS|r1| r| |d{VrdS|j |j|j||d{V}|jrd|_||_d|_dS||r|xjdz c_t'|jdzd |_t*d |j|j|j|j|jkrt1j|_d St*d |j||_d|_d |_d|_|d{Vd Sd S|j |j||j d{V}|jr|j!r%|j!|_t1j|_"nd |_d|_||_|j!s'||_d|_d|_|#dSd |_d S#tH$r&}t*%d|Yd}~d Sd}~wwxYw)a\Send or edit the streaming message. Returns True if the text was successfully delivered (sent or edited), False otherwise. Callers like the overflow split loop use this to decide whether to advance past the delivered chunk. ``finalize`` is True when this is the last edit in a streaming sequence. r(TN)r rrrrrjrg$@u@Flood control on edit (strike %d/%d), backoff interval → %.1fsFz0Edit failed (strikes=%d), entering fallback moderr[zStream send/edit error: %s)&rr*rrrqr0rer3r5r<rrrrr rr2r8rminr9rTrU_MAX_FLOOD_STRIKESrrr4rr7r6rrr#rr1rWrSr)r?rLrvisible_without_cursor_visible_stripped_MIN_NEW_MSG_CHARSrrs rrz#GatewayStreamConsumer._send_or_editYs$&&t,,"& 8? Y%;%C%CDHOUW%X%X "288::  4zz|| 4   $HO %HOt++)**-???4z +'R!t333 4%)%D4 $t!$ 99;;$#'"7"7"="=======$ $t#'<#<#< $ #'#3 $!) $=$$F ~,%-1*/3,./+#t  //77- //14//:= $ ;a ?;;D7#LL!= $ 3 $ 7 $ ;  $2T5LLL8<~7G7G 4',u  N /150D0D0F0F-481/4,-1*#44666666666$u!5 $|00 L !] 1   >!(5+1+<(48>3C3C00/4,)-D&+/D(!,9040D0D0F0F-481,9( ,,...4,1D( 5    LL5q 9 9 955555 s@-$L1LA LBL*A(LB7LL M"MM)NNN) rrr rr!r"r#r$r%r&)rAr)rArG)rLrrArG)rYrrArG)rLrrAr)rLrrrrAr)rAr)rrrAr)rLrrr rAr)rLrrAr)rLrrrrAr)%rrrrrrnrkr@propertyrDrFrKrPrWr\r^rarrrrcompiler staticmethodrrrrrrrrrrrrrrrrrr9s    26#'-1 1 1 1 1 1 f"""X")))X)&&&&1111 IIII@E###### $ $ $ $VVVVp$$$$L9L9L9L9b :;;I   \ &>////\ e#e#e#e#NYYYY BBBBB*4    ,3333jBGaaaaaaaarr)r __future__rrloggingr,rr dataclassesrtypingrr getLoggerrTobjectr`rJrNrrrrrrs$  #""""" !!!!!!  4 5 5 vxx fhh   + + + + + + +  + AAAAAAAAAAr