Bj d Z ddlmZ ddlZddlZddlZddlZddlZddlm Z ddl
m Z m Z m
Z
ddlmZ ddlmZ ddlmZmZmZ ej2 d Z e Z e Z e Z e G d
d Z G d d
Z y)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)AnyCallableOptional)BasePlatformAdapter)_custom_unit_to_cp) DEFAULT_STREAMING_EDIT_INTERVAL"DEFAULT_STREAMING_BUFFER_THRESHOLDDEFAULT_STREAMING_CURSORzgateway.stream_consumerc t e Zd ZU dZeZded< eZded< e Z
ded< dZ d ed
< d Z ded < d
Z
ded< dZded< y)StreamConsumerConfigz5Runtime config for a single stream consumer instance.float
edit_intervalintbuffer_thresholdstrcursorFbool buffer_only fresh_final_after_secondsauto transport chat_typeN)__name__
__module__ __qualname____doc__ _DEFAULT_STREAMING_EDIT_INTERVALr __annotations__#_DEFAULT_STREAMING_BUFFER_THRESHOLDr _DEFAULT_STREAMING_CURSORr r r r r zz z
z
z )zzz z z z
r r _draft_id_counterNc || _ || _ |xs
t | _ || _ || _ || _ t j | _ d| _
d | _ d | _ d| _
d| _ d| _ d| _ d| _ d| _ d| _ | j j( | _ d| _ t/ |dd du | _ d| _ d| _ d| _ d | _ d| _ y )Nr FTr r REQUIRES_EDIT_FINALIZE) adapterchat_idr cfgmetadata_on_new_message_initial_reply_to_idqueueQueue_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_use_draft_streaming _draft_id_draft_failures)selfr. r/ configr1 on_new_messageinitial_reply_to_ids r' __init__z GatewayStreamConsumer.__init__q s 3 1 3
. $7!#(;;= *. 59 " # " ! $)! " &*hh&<&<#$)!
G 5u = E
'
% %*!(, ! r& c | j S )z?True if at least one message was sent or edited during the run.)r: rJ s r' already_sentz"GatewayStreamConsumer.already_sent s !!!r& c | j S )zBTrue when the stream consumer delivered the final assistant reply.)rB rP s r' final_response_sentz)GatewayStreamConsumer.final_response_sent s (((r& c B | j j t y)z>Finalize the current stream segment and start a fresh message.N)r6 put _NEW_SEGMENTrP s r' on_segment_breakz&GatewayStreamConsumer.on_segment_break s %r& c L |r"| j j t |f yy)z7Queue a completed interim assistant commentary message.N)r6 rU _COMMENTARYrJ texts r'
on_commentaryz#GatewayStreamConsumer.on_commentary s KKOO[$ / 0 r& c ~ | j }|y | y# t $ r t j dd Y yw xY w)z8Fire the on_new_message callback, swallowing any errors.Nz on_new_message callback errorTexc_info)r2 Exceptionloggerdebug)rJ cbs r' _notify_new_messagez)GatewayStreamConsumer._notify_new_message sB
!
!
: I D I LL84L H Is <<Fpreserve_no_editc |r| j dk( ry d | _ d | _ d| _ d| _ d| _ d| _ | j r9t | xj dz
c_ t | j | _ y y )N __no_edit__r F )
r8 r9 r7 r= r> r? rG typer+ rH )rJ rf s r' _reset_segment_statez*GatewayStreamConsumer._reset_segment_state s{ 0 0M A #' ! $)! " $ $ J ( (A - ( !$Z 9 9DN %r& c f |r | j j | y|| j yy)u2 Thread-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)r6 rU rW rZ s r' on_deltaz GatewayStreamConsumer.on_delta s- KKOOD !
\ ! ! # r& c B | j j t y)z#Signal that the stream is complete.N)r6 rU _DONErP s r' finishz GatewayStreamConsumer.finish s r& c | j |z }d| _ |rJ| j rd}d}| j D ]1 }|j | }|dk7 s|dk( s||k s%|}t | }3 |rd| _ |||z d }nt d | j D }t | |kD r || d | _ y|| _ yd}d}| j D ] }d} |j || }|dk( r |dk( r+| j xs | j j d } nt|d| }
|
j d } | dk( r@| j xs | j j d xr |
j dk( } n|
| d z d j dk( } | r|dk( s||k r|}t | }|d z } |r(| xj |d| z
c_ d| _ |||z d }nd} | j D ]9 }t d t | D ] }
|j |d|
s|
| kD s |
} ! ; | r%| xj |d| z
c_ || d | _ y| xj |z
c_ y|rIyy)
ab Add 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 r FNc 3 2 K | ] }t | y w)N)len).0ts r' z?GatewayStreamConsumer._filter_and_accumulate.. s !IQ#a&!Is T
ri ) rF rE _CLOSE_THINK_TAGSfindrt max_OPEN_THINK_TAGSr7 endswithrfindstriprange)rJ r[ bufbest_idxbest_lentagidxmax_tag search_start is_boundary precedinglast_nl held_backis r' _filter_and_accumulatez,GatewayStreamConsumer._filter_and_accumulate s 4' ## 11 ,C ((3-Cbyh"nh#&#&s8 , +0D( h 1 23C "!I$2H2H!I IG;>s8g;MgXY D& TWD& 00 /C#$L !hhsL 9 "9 ! !8$($5$5 5 !D#'#4#4#=#=d#C (
),DS I&/ood&;G &"})-):):%: &I(,(9(9(B(B4(H%@(1(9R(? !, /8! .E.K.K.MQS.S &HNcHn'*H'*3xH !'*Qw 5 /< %%Yh 7%+/D( h 1 23C !"I # 4 4 .!&q#c(!3 .A "||CG 4Y ,- .. ! ))S9*-==)-0)-=* ))S0)W r& c | j r4| j s'| xj | j z
c_ d| _ yyy)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)rF rE r7 rP s r' _flush_think_bufferz)GatewayStreamConsumer._flush_think_bufferU s< d&:&: !3!3 3 !#D '; r& c
K t | j t r| j j nt }t | j dd }t
d| || j j z
dz
}| j | _
| j rct | xj dz
c_ t | j | _
t j d| j | j d}d}d } | j" j% }|t& u rd}nL|t( u rd}nAt |t* r t | d
k( r|d t, u r|d }n| j/ | r|r| j5 t7 j8 }|| j: z
} |xs |xs |d u}
| j j< sO|
xsK | | j> k\ xr | j@ xs, t | j@ | j jB k\ }
d} |
rL| j@ r? || j@ |kD r| jD | j jG | j@ || } d}
| jD xs | jH }| D ]' }| jK || d { }| ||k7 s&d}
) d
| _ d
| _& t7 j8 | _ |r|
| _' y |rd | _" d| _( d
| _)
|| j@ |kD r| jD | jT rtW | j@ || }| j@ jY dd | }||d
z k r|}| j@ d | }| j[ | d { }| jP s|sn_| j@ |d j] d | _ d | _" d
| _& || j@ |kD r| jD
| jT r| j@ }|s |s||| j j z
}| j[ ||xs | d { } t7 j8 | _ |r| j@ r| jP r$| j_ | j@ d { y | r| j` sd| _' y | jD r+| j[ | j@ d d { | _' y | jb s(| j[ | j@ d { | _' y |R| je | jg | d { t7 j8 | _ | je |rS| j@ r5| s3| jD r'| jD dk7 r| ji d { | je d tk jl d d { # t0 j2 $ r Y w xY w7 b7 {7 7 s7 .7 7 7 b7 5# tj jn $ rt d}| j@ rJ| jD r> tq | j[ | j@ d { 7 }n# tr $ r Y nw xY w|r| jN s d| _' Y y Y y Y y tr $ r }t ju d| Y d }~y d }~ww xY ww)z@Async task that drains the queue and edits the platform message.MAX_MESSAGE_LENGTH d ri zBStream consumer using native-draft transport (chat=%s draft_id=%s)TFN r len_fnr rx finalizerh re g?zStream consumer error: %s);
isinstancer. _BasePlatformAdaptermessage_len_fnrt rC r{ r0 r _resolve_draft_streamingrG rj r+ rH ra rb r/ r6
get_nowaitro rV tuplerY r r4 Emptyr time monotonicr< r rA r7 r r8 truncate_messager3 _send_new_chunkr= rB r> r? r; r r~
_send_or_editlstrip_send_fallback_finalrD r: rk _send_commentary#_flush_segment_tail_on_edit_failureasynciosleepCancelledErrorr r` error)rJ _len_fn
_raw_limit _safe_limitgot_donegot_segment_breakcommentary_textitemnowelapsed should_editcurrent_update_visiblechunkschunks_deliveredreply_tochunknew_id
_cp_budgetsplit_atok display_text_best_effort_okes r' runzGatewayStreamConsumer.run_ s $,,(<=
LL ' '
T\\+?F
#zGDHHOO,D Ds JK %)$A$A$C! $ $ J ( (A - ( !$Z 9 9DN LLT dnn
U 9 $)!"&
#{{ 5 5 7 5='+H ! </04 - !%dE2s4yA~$q'U`J`.21gO ! 33D9 & ,,. nn& 4 44 3(3&d2
xx++"- # D$?$?? 2 $ 1 1O t 0 01TXX5N5NN */& 4#4#4 1 12[@ , , 4 "& !>!> - -{7 "? " ,1(#'#3#3#Pt7P7P%+ 8E+/+?+?x+P%PF % 1f6H37 08 -/)/1,/3~~/?,#
9ID 5 ",/3D ,8=D 546D 1
1 12[@ , , 8 0 0%7 - -{G&
$(#4#4#:#:4J#O#kQ&66'2H $ 1 1)8 <#'#5#5e#< < 44B ",0,=,=hi,H,O,OPT,U)+/(/1,+ 1 12[@ , , 8 0 0* $(#4#4L#,=/BY$7 483E3E$"*"?.? 4F 4 .* ,0>>+;D(
(( 44"&";";DP>P $ 1 1D ?Q ? 9 D 5
"&!3!3>B>P>PQUQbQb>c8cD 5".--/ //@@@+/>>+;D(--/ % )) 6 , , , ,
= " F F HHH--t-D mmD)))A $ !;; ^ &QB =,. O 9 9d
AB I *%% 1 #O T%5%5&*1C1CDDUDU1V+V+V&WO t'@'@,0) (A 9 LL4a 8 8 9s C>[ X+
$W: .X+ /
W: 9X+ :.W: (X+ )W: :D8X+ 2X3X+ :X+ 4X+ 4[ 5B'X+ X A1X+ AX+ XAX+ *X +X+ /[ 0X+ [ ,X+ 2X 3 X+ <[ =*X+ 'X#( X+ 1[ 2&X+ X%A+X+ X'.X+ 3X)4X+ :X
X+ XX+ X+ X+ X+ X+ #X+ %X+ 'X+ )X+ +-[#Z<Y?
= Z[ Z[Z[+[ 1[9[[ [[ z[`"']?MEDIA:\s*\S+[`"']?c d| vrd| vr| S | j dd }t j j d| }t j dd| }|j S )u Strip 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)r[ cleaneds r' _clean_for_displayz(GatewayStreamConsumer._clean_for_displayX sa 4 $8$DK,, 3R8'1155b'B&&FG4~~ r& c 6 K | j | }|j s|S | j rt | j ni }| j j | j ||| d{ }|j rY|j rMt |j | _
d| _ || _ | j t |j S d| _ |S 7 r# t $ r"}t j# d| |cY d}~S d}~ww xY ww)zSend a new message chunk, optionally threaded to a previous message.
Returns the message_id so callers can thread subsequent chunks.
r/ contentr r1 NTFzStream send chunk error: %s)r r r1 dictr. sendr/ success
message_idr r8 r: r= rd r; r` ra r )rJ r[ reply_to_idmetaresultr s r' r z%GatewayStreamConsumer._send_new_chunkl s
&&t,zz| *.--4
&RD <<,, $ - F ~~&"3"3#&v'8'8#9 %)"'+$ ((*6,,--',$""!" LL6 : sT $DAC+ 6C)7A(C+ D C+ (D)C+ + D4D D DDDc | j xs d}| j j rH|j | j j r#|dt | j j }| j | S )z>Return the visible text already shown in the streamed message.r N)r= r0 r r} rt r rJ prefixs r' _visible_prefixz%GatewayStreamConsumer._visible_prefix s^ %%+ 88??vtxx ? 2c$((// 2 23F&&v..r& c | j xs | j }|r-|j | r |t | d j S |S )zAReturn only the part of final_text the user has not already seen.N)r? r
startswithrt r )rJ
final_textr s r' _continuation_textz(GatewayStreamConsumer._continuation_text sK &&@$*>*>*@ j++F3 c&kl+224 4r& c " || |k r| gS g }| } || |kD r^t ||| }|j dd| }||dz k r|}|j |d| ||d j d } || |kD r^|r|j | |S )z;Split text into reasonably sized chunks for fallback sends.rx r r N)r r~ appendr )r[ limitr r remainingr r s r' _split_text_chunksz(GatewayStreamConsumer._split_text_chunks s $<5 6M Y %'+IufEJ tQ
;H%1*$ MM)IX. /!(),33D9I
Y %' MM) $
r& c 8 K | j | }| j | }d| _ |j s
|j r|| j k7 r|}n| j
r| j r| j j r| j j | j j r{| j dt | j j } | j j | j | j
| d{ }|j r|| _ d| _ d| _ yt% | j dd }t' | j t( r| j j* nt }t- d|dz
}| j/ ||| } | j
}
d} d
} d}
| D ] }d}t1 d D ] }| j j3 | j || j4 d{ }|j r nL|d
k( rD| j7 | r3t8 j; d t= j> d d{ n |r |j sE|
r%d| _ d| _ | | _ | | _ d
| _ yd| _ d| _ d
| _ d
| _ yd}
|} |jB xs | } | jE ! |
r:|
| k7 r5t% | j dd }| || j |
d{ | | _ d| _ d| _ | d | _ d
| _ y7 0# t $ r Y 'w xY w7 O7 7 C# t $ r!}t8 j; d|
| Y d}~fd}~ww xY ww)zSend the final continuation after streaming edits stop working.
Retries each chunk once on flood-control failures with a short delay.
FNr/ r r Tr r r r r r r r/ r r1 r z.Flood control on fallback send, retrying in 3sg @delete_messagez(Fallback partial cleanup failed (%s): %srr )#r r r> r r r8 r= r0 r r} rt r. edit_messager/ r r` r: rB rC r r r r{ r r r r1 _is_flood_errorra rb r r r? r rd )rJ r[ r continuation
clean_textr raw_limitr
safe_limitr stale_message_idlast_message_idlast_successful_chunksent_any_chunkr attempt delete_fnr s r' r z*GatewayStreamConsumer._send_fallback_final s
,,T2
..z: $)!!!# !jD4H4H4J&J ) $$ ,, ,,55dhhooF!%!5!56LDHHOO8L7L!MJ '+||'@'@$(LL'+'7'7$. (A ( "
">>3=D 0 &*",0)DLL*>E $,,(<=
LL ' '
i#o.
(( z'(R ++)- " * 'EF 8
#|| 0 0 LL ! !]] 1
>>aEI$#DLL2BCCC + !$(! %bz " u" % . -L D LLB(! s C+N.5M #M$M ;C NM&AN-M).B"NM- 'M+(M- ,'NM M# N"M##N)N+M- - N6N
NNNc d t |dd xs d}|j }d|v xs
d|v xs d|v S )zFCheck if a SendResult failure is due to flood control / rate limiting.r r floodz retry afterrate)rC lower)rJ r err err_lowers r' r z%GatewayStreamConsumer._is_flood_error+ s? fgr*0bIIK )#X} 'AXVyEXXr& c | j j xs dj }|dk( ry|dk( ryt | j t
sy | j j
| j j xs d| j }|s;|d
k( r5t j d | j | j j yy# t $ r t j dd d}Y bw xY w) u Decide whether this run should use native draft streaming.
Honors ``cfg.transport``:
* ``"edit"`` → never use drafts (legacy progressive-edit path).
* ``"draft"`` → require draft support; gracefully fall back to edit
when the adapter declines. Logs the downgrade at debug.
* ``"auto"`` → use drafts when the adapter supports them for this
chat type; otherwise edit.
Adapter eligibility is checked via
:meth:`BasePlatformAdapter.supports_draft_streaming`, which considers
the chat type (e.g. Telegram drafts are DM-only) and platform-version
gates (e.g. python-telegram-bot 22.6+).
r editFoffN)r r1 z%supports_draft_streaming probe raisedTr^ draftuU Draft streaming requested but unsupported (chat=%s, type=%r) — falling back to edit)
r0 r r r r. r supports_draft_streamingr r1 r` ra rb r/ )rJ r supporteds r' r z.GatewayStreamConsumer._resolve_draft_streaming1 s XX''1688: $,,(<= == ((,,4 > I G# +LL$(("4"4
LL@4L P I s ?C "C32C3c K | j d| _ y | j j | j | j || j
d{ }t |dd s>t j dt |dd | xj dz
c_ d| _ y|| _ y
7 W# t $ r<}t j d| | xj dz
c_ d| _ Y d}~yd}~ww xY ww) a Emit a single animated draft frame for the current accumulated text.
Returns True when the frame landed. On any failure, permanently
disables drafts for the remainder of this run so subsequent frames
flow through the edit-based path (which can adapt with flood-control
backoff, etc.). Drafts have no message_id and clear naturally on
the client when the response finalizes via a regular sendMessage.
NF)r/ draft_idr r1 z=send_draft raised, disabling draft transport for this run: %sri r z@send_draft returned success=False, disabling draft transport: %sr unknownT) rH rG r.
send_draftr/ r1 r` ra rb rI rC r= rJ r[ r r s r' _send_draft_framez'GatewayStreamConsumer._send_draft_frame\ s >> ! ).D %
<<22 3 F vy%0 LLR3
A % (-D % # / LLOQR
A % (-D %
sB C:A B2 B0B2 AC:0B2 2 C7;2C2-C:2C77C:c H K | j s| j d{ | j xs | j }| j }|r-|j | r |t
| d j }| j | }|j sy | j j | j || j d{ }|j rd| _ yy7 7 # t $ r }t" j% d| Y d}~yd}~ww xY ww)aK Deliver 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.
Nr Tz"Segment-break tail flush error: %s)r> _try_strip_cursorr? r r7 r rt r r r r. r r/ r1 r r: r` ra r )rJ visibletailr r s r' r z9GatewayStreamConsumer._flush_segment_tail_on_edit_failure s ((((* * *''A4+?+?+A tw/G
&--/D&&t,zz| B <<,, - F
~~%)"
+ B LL=q A A BsL D"C2A?D"#5C6 C4C6 0D"4C6 6 D ?DD"D D"c 0 K | j r| j dk( ry| j }|r|j sy | j j | j
| j | d{ || _ y7 # t $ r Y yw xY ww)u Best-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.
rh Nr )r8 r r r. r r/ r= r` r s r' r z'GatewayStreamConsumer._try_strip_cursor s 4#3#3}#D %%'V\\^ ,,++ ++ ,
$*D
s< A B5B 8B9 B BB BBBBc p K | j | }|j sy | j j | j || j
d{ }|j r| j |j S 7 ,# t $ r }t j d| Y d}~yd}~ww xY ww)z6Send a completed interim assistant commentary message.Fr NzCommentary send error: %s) r r r. r r/ r1 r rd r` ra r r s r' r z&GatewayStreamConsumer._send_commentary s &&t,zz| <<,, - F ~~ ((*>> ! LL4a 8 s@ #B65B
B +B
B6B
B3B.)B6.B33B6c t | j dd xs d}|dk ry| j r| j dk( ry| j yt j
| j z
}||k\ S )a Return 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.
r r r Frh )rC r0 r8 r9 r r )rJ thresholdages r' _should_send_fresh_finalz.GatewayStreamConsumer._should_send_fresh_final sp DHH&A3GN3 > 4#3#3}#D # # +nn !9!99i r& c K | j } | j j | j || j d{ }t |dd sy|r:|dk7 r5t | j dd }| || j | d{ t |d d }|r!|| _ t j | _ nd| _ d| _ d
| _ || _
d
| _ y
7 # t
$ r }t j d| Y d}~yd}~ww xY w7 # t
$ r!}t j d|| Y d}~d}~ww xY ww) aC Send ``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.
r Nz1Fresh-final send failed, falling back to edit: %sFr rh r z+Fresh-final preview cleanup failed (%s): %sr T)r8 r. r r/ r1 r` ra rb rC r r r9 r: r= rB )rJ r[ old_message_idr r r new_message_ids r' _try_fresh_finalz&GatewayStreamConsumer._try_fresh_final sE )) <<,, - F vy%0 n
= .>EI$#DLL.AAA ! t< -D '+~~'7D $
-D '+D $ ! # $(!S
LLLa P B LLE& s}
E5C* C(C*
.E9D DD AE(C* * D3D EDED E!D=8E=EEr c K | j | }|}| j j r&|j | j j d }|j }|sy|j syd}| j
=| j j r'| j j |v rt
| |k ry| j r:|s8| j
,|| j k( ry| j | d{ }|ry | j
H| j r:|| j k( r|r | j sy|r*| j r| j | d{ ry| j j | j | j
|| d{ }|j" rd| _ t' |dd xs d}|rp|j( rd|j( | j
k7 rKt+ |j( | _ t- j. | _ d| _ | j3 n|| _ d| _ y| j7 | r| xj4 d z
c_ t9 | j: d
z d | _ t< j? d | j4 | j@ | j: | j4 | j@ k rt- j. | _! y
t< j? d| j4 | jE | _# d| _$ d
| _
d| _ | jK d{ y
y
| j jM | j || jN | jP d{ }|j" r|j( r+|j( | _ t- j. | _ nd
| _
d| _ || _ |j( s#| jE | _# d| _$ d| _ | j3 yd
| _
y
7 C7 7 7 7 # tR $ r } t< jU d| Y d} ~ y
d} ~ ww xY ww)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 T N)r/ r r r continuation_message_idsr% r ri r g $@u@ Flood control on edit (strike %d/%d), backoff interval → %.1fsFz0Edit failed (strikes=%d), entering fallback moder rh zStream send/edit error: %s)+r r0 r r r r8 rt rG r= r r; rD r r r. r r/ r r: rC r r r r r9 rd r@ r minrA ra rb _MAX_FLOOD_STRIKESr< r r? r> r r r3 r1 r` r )
rJ r[ r visible_without_cursor_visible_stripped_MIN_NEW_MSG_CHARSr r _continuation_idsr s
r' r z#GatewayStreamConsumer._send_or_edit s! &&t, "& 88??%;%C%CDHHOOUW%X " 2 8 8 : zz| $HHOOHHOOt+)*-??
% % ( t+++--d33B
R +'' t 3 33 T%D%D # ! 9 9 ;"&"7"7"= = = ##'<<#<#< $ #'#3#3 $!) $= $ F ~~-1* -4F> ((+1+<+<( 48>>3C0/4,)-D&+/D(!,,040D0D0F-481 ,9(
,,. ,1D( q 4F > N 7D LL5q 9 s C=Q;?Q Q;7Q ?Q; &Q &Q'Q +Q;,6Q "Q#B(Q Q; B,Q 8Q;9A Q Q Q Q; A Q Q
BQ 9Q;:Q Q;Q Q Q
Q Q8Q3.Q;3Q88Q;)NNNN) r. r r/ r rK z Optional[StreamConsumerConfig]r1 zOptional[dict]rL zOptional[callable]rM
Optional[str])returnr )r None)r[ r r r )rf r r r )r[ r r r )r[ r r r r r )r r )r r r r )r[ r r r r z'Callable[[str], int]'r z list[str])r[ r r r )r[ r r r r r )*r r r r r r| ry r+ r" rN propertyrQ rS rW r\ rd rk rm rp r r r r compiler staticmethodr r r r rt r r r r r r r r r r r r% r& r' r) r) M s
s 26#'-1-1A!A! A! / A!
! A! +
A! +A!F " " ) )&1
I @E :(
$ Vp$r9n
:;I & >/ *- &
({#zY )V&P BB*4 ,3j BG Vr& r) )!r
__future__r r loggingr4 r r dataclassesr typingr r r gateway.platforms.baser r r gateway.configr
r! r r# r r$ getLoggerra objectro rV rY r r) r% r& r' r+ s
# ! * * N 5
4 5 x h 8h hr&