(Gja N d Z ddlmZ ddlZddlZddlZddlZddlZddlZddl m
Z
ddl m Z m
Z
mZ ddlmZ ddlmZ ddlmZ dd lmZmZmZ dd
lm Z m Z ej d Z! e" Z# e" Z$ e" Z%e
G d d
Z& G d d Z'dS )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)MEDIA_TAG_CLEANUP_RE) DEFAULT_STREAMING_EDIT_INTERVAL"DEFAULT_STREAMING_BUFFER_THRESHOLDDEFAULT_STREAMING_CURSOR) is_intentional_silence_responseis_partial_silence_markerzgateway.stream_consumerc v e Zd ZU dZeZded< eZded< e Z
ded< dZ d ed
< d Z ded < d
Z
ded< dZded< dS )StreamConsumerConfigz5Runtime config for a single stream consumer instance.float
edit_intervalintbuffer_thresholdstrcursorFbool buffer_only fresh_final_after_secondsedit transport chat_typeN)__name__
__module__ __qualname____doc__ _DEFAULT_STREAMING_EDIT_INTERVALr __annotations__#_DEFAULT_STREAMING_BUFFER_THRESHOLDr _DEFAULT_STREAMING_CURSORr r r r r =/home/rurouni/.hermes/hermes-agent/gateway/stream_consumer.pyr r 6 s ??;M;;;; ?????+F++++ K (+ **** I Ir) r c e Zd ZU dZdZdZdZdZded< d\d]dZ dddd^d!Z
e d_d" Z e d_d# Z
e d`d% Ze d_d& Zdad(Zdd)dbd-Zdcd/Zdad0Zddd1Zdad2Zdd3ded5Zddd6Zdad7Zddd8Zedfd9 Zdad:Z dad;Z e Z e dfd< Z!dd=dgd?Z"dhd@Z#didBZ$e e%fdjdG Z&dddHZ'd_dIZ(d_dJZ)dcdKZ*dadLZ+dadMZ,dcdNZ-d_dOZ.dkdPZ/dldQZ0dmdSZ1dcdTZ2dUdVdndXZ3dadYZ4ddUdZdod[Z5dS )pGatewayStreamConsumera2 Async 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
r r _draft_id_counterNadapterr chat_idr config Optional[StreamConsumerConfig]metadataOptional[dict]on_new_messageOptional[callable]on_before_finalizeOptional[Callable[[], Any]]initial_reply_to_id
Optional[str]run_still_current Optional[Callable[[], bool]]c f || _ || _ |p
t | _ || _ || _ || _ || _ t j | _
d| _ d | _ d | _
t | _ d| _ d| _ d| _ d| _ d| _ d| _ d| _ d| _ d| _ | j j | _ d| _ d| _ g | _ t= |dd du | _ |pd | _ d| _! d| _" d| _# d | _$ d| _% d| _& d S )Nr FTr r REQUIRES_EDIT_FINALIZEc dS )NTr( r( r) r* z0GatewayStreamConsumer.__init__.. s r) )'r/ r0 r cfgr3 _on_new_message_on_before_finalize_initial_reply_to_idqueueQueue_queue _accumulated _message_id_message_created_tsset_preview_message_ids
_already_sent_edit_supported_last_edit_time_last_sent_text_last_edit_overflowed_fallback_final_send_fallback_prefix#_fallback_preserve_partial_messages_flood_strikesr _current_edit_interval_final_response_sent_final_content_delivered_delivered_commentary_textsgetattr_adapter_requires_finalize_run_still_current_in_think_block
_think_buffer_use_draft_streaming _draft_id_draft_failures_before_finalize_notified) selfr/ r0 r1 r3 r5 r7 r9 r; s r* __init__z GatewayStreamConsumer.__init__w s[ 3 1 3 3
. $6 $7!#(;== *. 59 14! " # " ! &+"$)! " 490 &*h&<#$)! ).%68(
G 5u = = E
' #4"E % %*!(, ! ).&&&r) Ffinal expect_editsrf r rg return dict | Nonec ` | j rt | j ni }|rd|d< |rd|d< |pdS )a7 Return per-send metadata for stream-created messages.
Mattermost treats notify-worthy sends as user-visible final content
when deciding whether a broken thread root may fall back flat. Preview
and progress sends keep their original metadata and remain thread-strict.
``expect_edits`` preserves the upstream Telegram streaming contract:
preview messages that may be edited later must stay on the editable
legacy send path, while fresh/fallback final sends can still use richer
final-message delivery.
Trg notifyN)r3 dict)rc rf rg metas r* _metadata_for_sendz(GatewayStreamConsumer._metadata_for_send sL " '+m;tDM""" (#'D " !DN|tr) c | j S )z?True if at least one message was sent or edited during the run.)rM rc s r* already_sentz"GatewayStreamConsumer.already_sent s
!!r) c | j S )zBTrue when the stream consumer delivered the final assistant reply.)rW rp s r* final_response_sentz)GatewayStreamConsumer.final_response_sent s
((r)
str | Nonec | j S )z?The Discord/chat message ID of the last-sent or edited message.)rI rp s r*
message_idz GatewayStreamConsumer.message_id s
r) c | j S )z|True when the final response content reached the user, even if
the subsequent cosmetic edit (cursor removal) failed.)rX rp s r* final_content_deliveredz-GatewayStreamConsumer.final_content_delivered s
,,r) Nonec K | j rdS d| _ | j dS | }t j | r
| d{V dS dS # t $ r Y dS w xY w)z?Run the pre-finalize hook exactly once, swallowing hook errors.NT)rb rC inspect isawaitable Exception)rc results r* _notify_before_finalizez-GatewayStreamConsumer._notify_before_finalize s ) F)-& # + F --//F"6**
DD s 0A
A A finalizerv contentr c N K | j ||d}||d< | j ru t j | j j j }d|v s+t d | D r
| j |d< n# t t f$ r Y nw xY w | j j di | d{V S )z>Edit via the adapter, passing routing metadata when supported.)r0 rv r r r3 c 3 F K | ] }|j t j j u V d S N)kindr{ Parameter VAR_KEYWORD).0params r* z6GatewayStreamConsumer._edit_message..( sC / / J'"3"??/ / / / / /r) Nr( ) r0 r3 r{ signaturer/ edit_message
parametersanyvalues TypeError
ValueError)rc rv r r kwargsparamss r*
_edit_messagez#GatewayStreamConsumer._edit_message s |$
&z =
*4<+DEEP ''3 / /!' / / / , ,' *. F:& z*
.T\.88888888888s A A8 8B B textc | |pd sdS | }|k rdS t fd| j D S )zDReturn True if *text* was already delivered as visible chat content.r FTc 3 H K | ] }| k V d S r )strip)r senttargets r* r z;GatewayStreamConsumer.has_delivered_text..9 s0 WWd4::<<6)WWWWWWr) )_clean_for_displayr _visible_prefixr rY )rc r visible_prefixr s @r* has_delivered_textz(GatewayStreamConsumer.has_delivered_text1 s ((44::<< 5 --//5577 V # #4WWWWd6VWWWWWWr) c D | j t dS )z>Finalize the current stream segment and start a fresh message.N)rG put _NEW_SEGMENTrp s r* on_segment_breakz&GatewayStreamConsumer.on_segment_break; s %%%%%r) c P |r#| j t |f dS dS )z7Queue a completed interim assistant commentary message.N)rG r _COMMENTARYrc r s r*
on_commentaryz#GatewayStreamConsumer.on_commentary? s3 1 KOO[$ / 0 0 0 0 0 1 1r) c | j }|dS | dS # t $ r t dd Y dS w xY w)z8Fire the on_new_message callback, swallowing any errors.Nz on_new_message callback errorTexc_info)rB r} loggerdebug)rc cbs r* _notify_new_messagez)GatewayStreamConsumer._notify_new_messageD sh
!
: F I BDDDDD I I I LL84L H H H H H H Is
&AApreserve_no_editr c |r
| j dk rd S d | _ d | _ d| _ d| _ d| _ d| _ d| _ d| _ d| _ | j r8t | xj dz
c_ t | j | _ d S d S )N __no_edit__r F )
rI rJ rH rP rR rS rT rW rX r_ typer. r` )rc r s r* _reset_segment_statez*GatewayStreamConsumer._reset_segment_stateN s 0M A A F #' ! $)! " 380
%*!(-% $ : JJ ( (A - ( ( !$ZZ 9DNNN : :r) c r |r | j | dS || dS dS )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)rG r r r s r* on_deltaz GatewayStreamConsumer.on_deltai sJ $ KOOD ! ! ! ! !
\ ! ! # # # # # \r) c D | j t dS )z#Signal that the stream is complete.N)rG r _DONErp s r* finishz GatewayStreamConsumer.finishu s r) c | j |z }d| _ |r| }| j rd}d}| j D ]L}| | }|dk r |dk s||k r|}t | }M|rd| _ |||z d }n@t
d | j D }t | |k r || d n|| _ dS d}d}| j D ] }| } d}
| | |
}|dk rn|dk r#| j p| j d } n|d| } |
d }
|
dk r;| j s| j d o| dk } n#| |
d z d dk } | r |dk s||k r|}t | }n|d z }
|r-| xj |d| z
c_ d| _ |||z d }nd}| j D ][}| } t d t | D ]'}| | d| r||k r|}(\|r*| xj |d| z
c_ || d | _ n#| xj |
| z
c_ dS |dS dS )
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 4 K | ]}t | V d S r )len)r ts r* r z?GatewayStreamConsumer._filter_and_accumulate.. s( !I!IQ#a&&!I!I!I!I!I!Ir) T
r )r^ lowerr] _CLOSE_THINK_TAGSfindr max_OPEN_THINK_TAGSrH endswithrfindr range_strip_orphan_close_tags)rc r buf lower_bufbest_idxbest_lentagidxmax_tag tag_lower search_start is_boundary precedinglast_nl held_backis r* _filter_and_accumulatez,GatewayStreamConsumer._filter_and_accumulate s 4' X
I# R
1 , ,C#..55Cbyyh"nnh#&#&s88 +0D( h 1 2 23CC "!I!I$2H!I!I!I I IG;>s88g;M;MgXYY SVD&F 0 / /C # I#$L/ 'nnY E E "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 %%YhY 7%%+/D( h 1 2 23CC !"I # 4 . .$'IIKK !&q#c((!3!3 . .A ( 1 1)BQB- @ @ .Q]],- . !
P ))S9*-==))-0)-=** ))T-J-J3-O-OO))Fq X X X X X r) c . d|vr|S | }g }d}|t | k rd}|||dz dk r| j D ]}| }t | }||||z |k rJ||z } | t | k r,|| dv r"| dz
} | t | k r
|| dv "| }d} n|s | || |dz
}|t | k d | S ) u Remove any close tags from *text* that have no matching open.
Mirrors ``agent/think_scrubber.py::StreamingThinkScrubber.
_strip_orphan_close_tags`` so the progressive-display filter
behaves the same as the post-stream final-response scrubber.
An orphan close tag is always noise — stripped along with any
trailing whitespace so surrounding prose flows naturally.
z
47###Q #d))mm wws||r) c | j r8| j s3| xj | | j z
c_ d| _ dS dS dS )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] rH r rp s r* _flush_think_bufferz)GatewayStreamConsumer._flush_think_buffer sb $d&: $
!>!>t?Q!R!R R !#D $ $ $ $r) c K t | j t r | j j nt }| }t
d| || j j z
dz
}| | _
| j
r\t | xj dz
c_ t | j | _
t d| j | j
| sdS d}d}d} | j }|t( u rd}nx|t* u rd}nlt |t, r+t | dk r|d t. u r |d }n,| | n# t2 j $ r Y nw xY w|rW| t9 | | j r | d{V dS tA j! }|| j" z
} |p|p|du}
| j j# s6|
p3| | j$ k r| j p!t | j | j j% k }
d} |
r/|s-|s+|)tM | | j rd}
|
r| j r || j |k r| j' | j ( | j ||
} d}
| j' p| j) }| D ]*}| * ||| d{V }|||k rd}
+d | _ d | _+ tA j! | _" |r|
| _, |
rd| _- dS |rd| _' d| _. d | _/ || j |k r| j' | j0 rtc | j || }| j 2 d
d | }||dz k r|}| j d| }| 3 |dd d{V }| j. s|snW| j |d 4 d
| _ d| _' d | _+ || j |k r| j' | j0 | j }|s|s||| j j z
}| 3 ||p|| d{V } tA j! | _" |r-| j s| j' | j5 r| 6 d{V | j r| j. r!| 7 | j d{V n| j, rd| _- n| r | j8 r| j9 rd| _, d| _- n| j' r^| 3 | j d d{V | _, | j, rd| _- nb| j. r | 7 | j d{V n:| j5 s3| 3 | j d{V | _, | j, rd| _- dS |[| : | ; | d{V tA j! | _" | : |rK| j r.| s,| j' r%| j' dk r| < d{V | : d t{ j> d d{V # tz j? $ rs d}| j rI| j' rB t | 3 | j dd d{V }n# t $ r Y nw xY w|r| j, sd| _, d| _- Y dS Y dS Y dS t $ r&}t B d| Y d}~dS d}~ww xY w)z@Async task that drains the queue and edits the platform message. d r zBStream consumer using native-draft transport (chat=%s draft_id=%s)TNFr r len_fnrf r r r
is_turn_finalr r r g?zStream consumer error: %s)C
isinstancer/ _BasePlatformAdaptermessage_len_fnr _raw_message_limitr rA r _resolve_draft_streamingr_ r r. r` r r r0 r\ rG
get_nowaitr r tupler r rE Emptyr _is_intentional_silence_responser rH _suppress_silence_markertime monotonicrO r rV r _is_partial_silence_markerrI truncate_messagerD _send_new_chunkrP rW rX rR rS rN r r
_send_or_editlstriprM r _send_fallback_finalr[ rQ r _send_commentary#_flush_segment_tail_on_edit_failureasynciosleepCancelledErrorr r} error)rc _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 sW
$,(<==
DL ' ' ,,..
#zGGDHO,D,D Ds JKK %)$A$A$C$C! $ JJ ( (A - ( ( !$ZZ 9DN LLT dn
~ 9^
* ..00 F !$)!"&
#{ 5 5 7 7 5=='+H ! <//04 - !%dE22 "s4yyA~~$q'U`J`J`.21gO ! 33D9999 ; & ,,... 8 //0ABB # ; ; = ======== n&& 44 3(3&d2
x+ "- # D$?? 2 $ 1O t 011TX5NN */& ( $ ( . ( (/2 //0ABB 0
#(K b<4#4 b< 122[@@ , 4 "& !>!> -{7 "? " " ,1(#'#3#Pt7P%+ 8 8E+/+?+? % (&. ,@ , , & & & & & & F
& 1f6H6H37 0,.)/1,/3~/?/?,# #
9ID 5 / E@D = "F, 7/3D ,8=D 546D 1
122[@@ , 8 0 9 &8 -{G& &
$(#4#:#:4J#O#O#kQ&666'2H $ 1)8) < $(#5#5 !D $6 $ $ 4 "B " ",0,=hii,H,O,OPT,U,U)+/(/1,C 122[@@ , 8 0 9D $(#4L# 8,= 8/BY$7 483E3E$"*"?.? '/ 4F 4 4 . . . . . .* ,0>+;+;D( ;( =D,<,HDL^,H " : : < <<<<<<<<
( 3E 4 2E"&";";DP>P $ 1D ?Q ? ? 9 9 9 9 9 9 D 5 $ 8
S@D = =!%!: S '+&?&?@Q&R&R R R R R R R R!%!3 E>B>P>PQUQb>c>c8c8c8c8c8c8cD 5 # 8 E@D =F".--/// //@@@@@@@@@+/>+;+;D(--/// % E )I 6I ! ,I ! ,
= = " F F H HHHHHHHH--t-DDD mD)))))))))}^
*@ % 5 5 5 $O
T%5
&* " 0 0 -E 1 ' 'OO
! D
5t'@
5,0)04----
5
5
5
5
5
5 9 9 9 LL4a 8 8 8 8 8 8 8 8 8 9s *Z( Z( $F -Z( . F 9Z( :?F 9Z( :F Z( F" Z( !F""AZ( ?EZ( JZ( "CZ( ( ]0[98]9
\]\]) ]2]]c * t 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.
)r "strip_media_directives_for_display)r s r* r z(GatewayStreamConsumer._clean_for_display{ s $FtLLLr) r reply_to_idc 8 K | | }| s|S | j | j ||| |d d{V }|j rk|j rdt |j | _ |
| d| _ || _ |
t |j S d| _ |S # t $ r'}t d| |cY d}~S d}~ww xY w)zSend a new message chunk, optionally threaded to a previous message.
Returns the message_id so callers can thread subsequent chunks.
Tre r0 r r r3 NFzStream send chunk error: %s)r r r/ sendr0 rn successrv r rI _track_preview_ids_from_resultrM rP r rN r} r r )rc r r rf r~ r s r* r z%GatewayStreamConsumer._send_new_chunk sZ &&t,,zz|| <,, $ 00u40PP - F ~
#&"3
##&v'8#9#9 33F;;;%)"'+$ ((***6,---',$"" LL6 : : : s$ B/C( C( (
D2 DDDc | j pd}| j j rA| | j j r"|dt | j j }| | S )z>Return the visible text already shown in the streamed message.r N)rP rA r r r r rc prefixs r* r z%GatewayStreamConsumer._visible_prefix se %+ 8? 4vtx ? ? 4 2c$(/ 2 2 2 23F&&v...r)
final_textc | j p| }|r>| | r)|t | d S |S )zAReturn only the part of final_text the user has not already seen.N)rS r
startswithr r )rc r' r& s r* _continuation_textz(GatewayStreamConsumer._continuation_text s_ &@$*>*>*@*@ 5j++F33 5 c&kkll+2244 4r) limitr 'Callable[[str], int]' list[str]c v || |k r| gS g }| } || |k r|t ||| }| dd| }||dz k r|}| |d| ||d d } || |k ||r| | |S )z;Split text into reasonably sized chunks for fallback sends.r r r N)r r r r )r r+ r r remainingr r s r* _split_text_chunksz(GatewayStreamConsumer._split_text_chunks s 6$<<5 6M fY %''+IufEEJ tQ
;;H%1*$$ MM)IXI. / / /!()),33D99I
fY %'' % MM) $ $ $
r) c K | | }| | }d| _ | s| r|| k r|}n| j r| j r| j j r| j | j j ri| j dt | j j } | | j | d{V }|j r|| _ n# t $ r Y nw xY wd| _ d| _ d| _ dS t# | j dd }t' | j t( r | j j nt }t- d|dz
}| ||| } | j }
d} d
} d}
| D ]}d}t1 d D ]}| j | j || d
d{V }|j r nQ|dk rJ| | r5t: d t? j d d{V |r|j s@|
r d| _ | | _ | | _ d
| _! dS d| _ d| _ d
| _ d
| _! dS d}
|} |j" p| } | # |
rw|
| k rq| j$ sj||k rdt# | j dd }|L || j |
d{V n3# t $ r&}t: d|
| Y d}~nd}~ww xY w| | _ d| _ d| _ 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.
FNrv r TMAX_MESSAGE_LENGTH r r r r r r r0 r r3 r z.Flood control on fallback send, retrying in 3sg @delete_messagez(Fallback partial cleanup failed (%s): %sr )%r r* rR r r rI rP rA r r r r r" r} rM rW rX rZ r/ r r r r r0 r r! r0 rn _is_flood_errorr r r r rS rv r rT )rc 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
,,T22
..z:: $)!!!## $ !!
jD4H4H4J4J&J&J ) $ , ,55dhoFF "&!56LDHO8L8L7L6L!MJ '+'9'9'+'7$. (: ( ( " " " " " " "> >3=D 0$ %)",0)04-DL*>EE $,(<==
DL ' '
i#o..
(( z'(RR +)- " , ' , 'EF 88
#| 0 0 L ! ! 4 44 4 @ @ 1
> Ea<EEI$#)DL2BCCCCCCCCCC LLB(! + !$(!(,% %bz " 38000s* 0D
DD8L
M L;;M c h t |dd pd}| }d|v pd|v pd|v S )zFCheck if a SendResult failure is due to flood control / rate limiting.r r floodz retry afterrate)rZ r )rc r~ err err_lowers r* r7 z%GatewayStreamConsumer._is_flood_error[ sD fgr**0bIIKK )#X} 'AXVyEXXr) c | j j pd }|dk rdS |dk rdS t | j t
sdS | j | j j pd| j }n.# t $ r! t dd d}Y nw xY w|s3|d k r+t d
| j | j j dS dS ) 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 FoffN)r r3 z%supports_draft_streaming probe raisedTr draftuU Draft streaming requested but unsupported (chat=%s, type=%r) — falling back to edit)
rA r r r r/ r supports_draft_streamingr r3 r} r r r0 )rc r supporteds r* r z.GatewayStreamConsumer._resolve_draft_streaminga s X'1688:: 5 5 $,(<== 5 == (,4 > II LL@4L P P P III G## +L$("4
5ts -A< <(B'&B'c K | j d| _ dS | j | j | j || j d{V }nJ# t $ r=}t d| | xj dz
c_ d| _ Y d}~dS d}~ww xY wt |dd sCt dt |dd | xj dz
c_ d| _ dS || _ d
S ) 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)r0 draft_idr r3 z=send_draft raised, disabling draft transport for this run: %sr r" z@send_draft returned success=False, disabling draft transport: %sr unknownT) r` r_ r/
send_draftr0 r3 r} r r ra rZ rP rc r r~ r s r* _send_draft_framez'GatewayStreamConsumer._send_draft_frame sH > ! ).D %5
<22 3 FF LLOQR
A % (-D %55555
vy%00 LLR33
A % (-D %5 # ts 3A
B2B
Bc H K | j s| d{V | j p| }| j }|r>| | r)|t
| d }| | }| sdS | j
| j || j
d{V }|j r d| _ dS dS # t $ r&}t" d| Y d}~dS d}~ww xY w)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.
Nr5 Tz"Segment-break tail flush error: %s)rR _try_strip_cursorrS r rH r) r r r r r/ r! r0 r3 r" rM r} r r )rc visibletailr~ r s r* r z9GatewayStreamConsumer._flush_segment_tail_on_edit_failure sv ( +((** * * * * * * *'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 Bs 2;C1 1
D!;D D!c K | j r | j dk rdS | }|r| sdS | | j | d{V || _ dS # t
$ r Y dS w xY w)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.
r Nr2 )rI r r r rP r} r% s r* rS z'GatewayStreamConsumer._try_strip_cursor s 4#3}#D#D F%%'' V\\^^ F $$ + %
$*D DD s )A/ /
A=<A=c K | | }| sdS | j | j || j d{V }|j r.| | j | |j S # t $ r&}t d| Y d}~dS d}~ww xY w)z6Send a completed interim assistant commentary message.Fr5 NzCommentary send error: %s)
r r r/ r! r0 r3 r" r rY r r} r r rP s r* r z&GatewayStreamConsumer._send_commentary s &&t,,zz|| 5 <,, - F ~
> ((*** 077===> ! LL4a 8 8 855555 s A(B
C"CCc t | j dd pd}|dk rdS | j r | j dk rdS | j dS t 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 Fr )rZ rA rI rJ r r )rc thresholdages r* _should_send_fresh_finalz.GatewayStreamConsumer._should_send_fresh_final sv DH&A3GGN3 >>5 4#3}#D#D5 # +5n !99i r) c > t | j dd }t | j t rl | j }n4# t
$ r'}t d| d}Y d}~nd}~ww xY wt |t r||k r|S |S )a Per-message length budget (in the adapter's ``message_len_fn`` units)
before the consumer splits an overflowing reply.
Adapters with a richer send/draft path (e.g. Telegram rich messages)
can raise this above ``MAX_MESSAGE_LENGTH`` via
``streaming_overflow_limit`` so a reply that fits one rich message isn't
fragmented at the legacy edit limit. Falls back to
``MAX_MESSAGE_LENGTH`` (4096 default) for everyone else.
r3 r4 z)streaming_overflow_limit check failed: %sN) rZ r/ r r streaming_overflow_limitr} r r r )rc basecapr s r* r z(GatewayStreamConsumer._raw_message_limit s t|%94@@ dl$8 9 9
l;;==
H!LLL
#s##
d
s A
A= A88A=c l |r/|dk r+| j t | dS dS dS )z9Record a real preview message id for fresh-final cleanup.r N)rL addr )rc rv s r* _track_preview_idz'GatewayStreamConsumer._track_preview_id4 sG ;*
55 % ) )#j// : : : : : ; ;55r) r~ c V | t |dd t |dd pdD ]}| | t |dd pi }t |t r/| d pdD ]}| | dS dS )zRecord every message id a send/edit result exposes: the primary id
plus any continuation ids from an oversized split
(``continuation_message_ids`` or ``raw_response['message_ids']``).rv Ncontinuation_message_idsr( raw_response message_ids)rb rZ r rl get)rc r~ midraws r* r# z4GatewayStreamConsumer._track_preview_ids_from_result9 s
wv|T B BCCCF$>EEK ( (C " "3 ' ' ' 'fnd339r c4 ,
..4"
,
,&&s++++ , ,
,
,r) c . | j r | j dk rdS t | j dd }|dS ||| j }n# t $ r || }Y nw xY wn3# t
$ r&}t d| Y d}~dS d}~ww xY w|du S )u' Return True when the adapter would rather finalize a streamed reply
by sending a fresh message and deleting the preview than by editing the
preview in place — e.g. Telegram, whose ``sendRichMessage`` send path
currently renders richer markdown than Hermes' MarkdownV2 edit path.
Returns False when there is no real preview to replace (no message id,
or the ``__no_edit__`` sentinel), when the adapter doesn't expose the
hook, or on any error (the consumer then keeps the edit-in-place path).
r F prefers_fresh_final_streamingN)r3 z.prefers_fresh_final_streaming check failed: %sT)rI rZ r/ r3 r r} r r )rc r fnr~ r s r* _adapter_prefers_fresh_finalz2GatewayStreamConsumer._adapter_prefers_fresh_finalE s 4#3}#D#D5
T\#BD
I
I
:5
"D4=999
"
"
" D
" LLI1 M M M55555 ~ s5 A A A A A A
B*B BTr r c V K t | j }| j r%| j dk r| | j | j | j || d d{V }n3# t $ r&}t
d| Y d}~dS d}~ww xY wt |dd sdS t |d d }t | j d
d }|_|D ]\}|r |dk s||k r || j | d{V *# t $ r&}t
d || Y d}~Ud}~ww xY wt | _ |r || _ t j
| _ nd| _ d| _ d| _ || _ |rd| _ dS ) a_ 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.
``is_turn_final`` is False when finalizing an interim segment at a tool
boundary (a preamble) rather than the turn-final answer; the
final-delivery flag is then left unset so the gateway still delivers the
real answer from the next API call (#29346).
Ported from openclaw/openclaw#72038.
r Tr r5 Nz1Fresh-final send failed, falling back to edit: %sFr" rv r6 z+Fresh-final preview cleanup failed (%s): %s)rK rL rI ra r/ r! r0 rn r} r r rZ r r rJ rM rP rW ) rc r r stale_idsr~ r new_message_idrA stale_ids r* _try_fresh_finalz&GatewayStreamConsumer._try_fresh_finalc sI " 122 , 0M A A MM$* + + + <,, 00t0<< - FF
LLLa P P P55555 vy%00 5 ! t<< DL*:DAA %
8}#<#<N@Z@Z #)DL(;;;;;;;;;; LLE !
%(EE! , -D '+~'7'7D $ $
-D '+D $ ! # -(,D %ts0 > =AD 9'.v~t'L'L %lD99 )l>N>NOa>b>b ) 03 , 0 01B C C !4#)#4!4#'#30 0 D ,
0 ! (
5+1+<( 48>3C3C0 ;;FCCCC/4,)-D&+/D(!, 9040D0D0F0F-481 ,9(
,,... 4 ,1D( 5 LL5q 9 9 955555 sH 0$W BW CW 4E2W (BW A(W ,C"W W
X #XX )NNNNNN)r/ r r0 r r1 r2 r3 r4 r5 r6 r7 r8 r9 r: r; r< )rf r rg r rh ri )rh r )rh rt )rh ry )rv r r r r r )r r rh r )r r rh ry )r r rh ry )r r rh r )r r r r: rf r rh r: )rh r )r' r rh r )r r r+ r r r, rh r- )rh r )rv r: rh ry )r~ r rh ry )r r r r rh r )r r r r r r rh r )6r r! r" r# r~ r r r. r% rd rn propertyrq rs rv rx r r r r r r r r r r classmethodr r r r
_MEDIA_RE staticmethodr r r r* r r0 r r7 r rQ r rS r r[ r rb r# rm rs r r r( r) r* r, r, S su
26#'-1:>-1:>`/ `/ `/ `/ `/J " 0 " " " X" ) ) ) X) X - - - X-
& 9 9 9 9 9 9:X X X X& & & &1 1 1 1
I I I I @E : : : : : :6
$
$
$
$ c c c cJ [ @
$
$
$
$^9 ^9 ^9 ^9L %I
M
M
M \
M" # # # # # # J/ / / / *- \(I9 I9 I9 I9VY Y Y Y ) ) ) )V& & & &P B B B BB ( < , .; ; ; ;
,
,
,
, < JN C C C C C CJ)
)
)
)
X .3$u u u u u u u ur) r, )(r#
__future__r r r{ loggingrE rer dataclassesr typingr r r gateway.platforms.baser r r r
gateway.configr r$ r r& r
r' gateway.response_filtersr r r r getLoggerr objectr r r r r, r( r) r* r s
# " " " " " ! ! ! ! ! ! * * * * * * * * * * N N N N N N 5 5 5 5 5 5 7 7 7 7 7 7
4 5 5 vxx fhh 8u u u u u u u u u ur)