ei;dZddlmZddlZddlZddlZddlZddlZddlm 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_consumercLeZdZUdZdZded<dZded<dZd ed <d Zd ed <dS)StreamConsumerConfigz5Runtime config for a single stream consumer instance.g?float edit_interval(intbuffer_thresholdu ▉strcursorFbool buffer_onlyN) __name__ __module__ __qualname____doc__r __annotations__r rr9/home/div/.hermes/hermes-agent/gateway/stream_consumer.pyrr(s]??MFKrrc@eZdZdZdZdZdZ d4d5dZed6dZ ed6dZ d7dZ d8dZ ddd9dZ d8dZd7dZd8dZd7dZd7dZejd Zed:d!Zd;d$Zdd+Zd8d,Zd6d-Zd7d.Zd7d/Zd?d0Z dd1d@d3Z!dS)AGatewayStreamConsumera2Async consumer that progressively edits a platform message with streamed tokens. Usage:: consumer = GatewayStreamConsumer(adapter, chat_id, config, metadata=metadata) # Pass consumer.on_delta as stream_delta_callback to AIAgent agent = AIAgent(..., stream_delta_callback=consumer.on_delta) # Start the consumer as an asyncio task task = asyncio.create_task(consumer.run()) # ... run agent in thread pool ... consumer.finish() # signal completion await task # wait for final edit )zzz z z z )zzz z z z Nadapterrchat_idrconfigOptional[StreamConsumerConfig]metadataOptional[dict]c||_||_|p t|_||_t j|_d|_d|_ d|_ d|_ d|_ d|_ d|_d|_d|_|jj|_d|_t)|dddu|_d|_d|_dS)NFTgrREQUIRES_EDIT_FINALIZE)rrrcfgr!queueQueue_queue _accumulated _message_id _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)selfrrrr!s r__init__zGatewayStreamConsumer.__init__Ps  3133  #(;== *."#"!$)! "&*h&<#$)! G5u = = E ' %rreturnrc|jS)z?True if at least one message was sent or edited during the run.)r,r9s r already_sentz"GatewayStreamConsumer.already_sentts !!rc|jS)zBTrue when the stream consumer delivered the final assistant reply.)r4r=s rfinal_response_sentz)GatewayStreamConsumer.final_response_sentys ((rNonecD|jtdS)z>Finalize the current stream segment and start a fresh message.N)r)put _NEW_SEGMENTr=s ron_segment_breakz&GatewayStreamConsumer.on_segment_break~s  %%%%%rtextcP|r#|jt|fdSdS)z7Queue a completed interim assistant commentary message.N)r)rC _COMMENTARYr9rFs r on_commentaryz#GatewayStreamConsumer.on_commentarys3  1 KOO[$/ 0 0 0 0 0 1 1rFpreserve_no_editrLcj|r |jdkrdSd|_d|_d|_d|_d|_dS)N __no_edit__r$F)r+r*r/r0r1)r9rLs r_reset_segment_statez*GatewayStreamConsumer._reset_segment_statesJ   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)rCrErIs ron_deltazGatewayStreamConsumer.on_deltasJ  $ KOOD ! ! ! ! ! \  ! ! # # # # #\rcD|jtdS)z#Signal that the stream is complete.N)r)rC_DONEr=s 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 ) r8r7_CLOSE_THINK_TAGSfindrXmax_OPEN_THINK_TAGSr*endswithrfindstriprange)r9rFbufbest_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)r8r7r*r=s r_flush_think_bufferz)GatewayStreamConsumer._flush_think_buffersR   $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_LENGTHdTFNrr]r$r\finalizerNrKg?zStream consumer error: %s)0r5rr`rXr&rr) get_nowaitrSrD isinstancetuplerHrrr'Emptyrttime monotonicr.rr3r*r r+truncate_message_send_new_chunkr/r,r4r0r1r-rc _send_or_editlstrip_send_fallback_finalr6rO_send_commentary#_flush_segment_tail_on_edit_failureasynciosleepCancelledErrorr Exceptionloggererror)r9 _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)rFcleaneds 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}|jrB|jr;t|j|_ d|_ ||_ t|jSd|_ |S#t$r'}td||cYd}~Sd}~wwxYw)zSend a new message chunk, optionally threaded to a previous message. Returns the message_id so callers can thread subsequent chunks. )rcontentreply_tor!NTFzStream send chunk error: %s)rrdr!dictrsendrsuccess message_idrr+r,r/r-rrr)r9rFrmetaresultrs rrz%GatewayStreamConsumer._send_new_chunks: &&t,,zz||   *.-?4 &&&RD<,, $ -F ~ #&"3 ##&v'8#9#9 %)"'+$6,---',$""    LL6 : : :       s$BC>C C8C3-C83C8c|jpd}|jjrA||jjr"|dt |jj }||S)z>Return the visible text already shown in the streamed message.r$N)r/r&rrbrXrr9prefixs r_visible_prefixz%GatewayStreamConsumer._visible_prefix se%+ 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)r1r startswithrXr)r9rrs r_continuation_textz(GatewayStreamConsumer._continuation_texts_&@$*>*>*@*@  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.r\rrzN)rXrcappendr)rFrr remainingrs r_split_text_chunksz(GatewayStreamConsumer._split_text_chunkss t99  6M )nnu$$ tQ66H%1*$$  MM)IXI. / / /!()),33D99I )nnu$$  % MM) $ $ $ rcJK||}||}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. FNrrrTrvrwrxryr$rzrrr!rz.Flood control on fallback send, retrying in 3sg@rV)rrr0rdrr+r/r&rrbrXr edit_messagerrrr,r4r5r`rrerr!_is_flood_errorrdebugrrr1r)r9rFr continuation clean_textr raw_limit safe_limitrlast_message_idlast_successful_chunksent_any_chunkrattempts rrz*GatewayStreamConsumer._send_fallback_final,sj ,,T22 ..z:: $)!!!##$ !! jD4H4H4J4J&J&J) $,,55dhoFF  "&!56LDHO8L8L7L6L!MJ '+|'@'@$(L'+'7$.(A(("""""" ">>3=D0$%)",0)DL*>EE i#o.. ((zBB)- "' C' CEF 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)r0_try_strip_cursorr1rr*rrXrrrdrrrr!rr,rrr)r9visibletailrrs 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. rNNr)r+rrdrrrr/rrs rrz'GatewayStreamConsumer._try_strip_cursors  4#3}#D#D F%%'' V\\^^  F ,++ +,        $*D     DD s4A:: BBc,K||}|sdS |j|j||jd{V}|jS#t$r&}t d|Yd}~dSd}~wwxYw)z6Send a completed interim assistant commentary message.FrNzCommentary send error: %s) rrdrrrr!rrrr)r9rFrrs rrz&GatewayStreamConsumer._send_commentarys&&t,,zz|| 5 <,, -F> !    LL4a 8 8 855555 s3A## B-BBr{r|cK||}|}|jjr ||jjd}|}|sdS|sdSd}|j/|jjr#|jj|vrt ||krdS |jn|jrd||jkr |r|j sdS|j |j |j||d{V}|j rd|_||_d|_dS||r|xjdz c_t#|jdzd |_t&d |j|j|j|j|jkrt-j|_d St&d |j||_d|_d |_d|_|d{Vd Sd S|j |j ||j d{V}|j rY|jr |j|_nd |_d|_||_|js'||_d|_d|_dSd |_d S#t@$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)rrrr|rr]rzg$@u@Flood control on edit (strike %d/%d), backoff interval → %.1fsFz0Edit failed (strikes=%d), entering fallback moderrNzStream send/edit error: %s)"rr&rrrdr+rXr-r/r6rrrrr,r2rminr3rr_MAX_FLOOD_STRIKESrrr.rr1r0rrr!rrr)r9rFr|visible_without_cursor_visible_stripped_MIN_NEW_MSG_CHARSrrs rrz#GatewayStreamConsumer._send_or_edits&&t,,"& 8? Y%;%C%CDHOUW%X%X "288::  4zz|| 4   $HO %HOt++)**-???4a +'B!t333 4%)%D4 $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+<((/4,)-D&+/D(!,9040D0D0F0F-481,9(4,1D( 5    LL5q 9 9 955555 s:-$J9A J9BJ97A(J9#B J90J99 K)K$$K))NN)rrrrrr r!r")r;r)r;rA)rFrr;rA)rLrr;rA)rFrr;r)rFrrrr;r)r;r)rrr;r)rFrrr r;r)rFrr;r)rFrr|rr;r)"rrrrrrar^r:propertyr>r@rErJrOrQrTrrrtrrcompiler staticmethodrrrrrrrrrrrrrrrr1s    26#' " " " " " H"""X")))X)&&&&1111 @E###### $ $ $ $VVVVp$$$$L9L9L9L9b :;;I   \ &8////\ b#b#b#b#HYYYY BBBBB**BGHHHHHHHHrr)r __future__rrloggingr'rr dataclassesrtypingrr getLoggerrobjectrSrDrHrrrrrrs$  #""""" !!!!!!  4 5 5 vxx fhh   x x x x x x x x x x r