+ jHj<$Rt^RIHt^RIt^RIt^RIt^RIt^RIt^RI t ^RI t ^RI t ^RI t ^RI t ^RIHt^RIHtHtHt^RIHt^RIHt^RIHtHtHt^RIH t!H"t#^R I$H%t%^R I&H't'^R I(H)t)^R I*H+t+]PX!R 4t-].!4t/].!4t0].!4t1].!4t2].!4t3].!4t4].!4t5].!4t6]Pn]PpPn3t9Rt:]!RR44t;]!RR44t<!RR]'])]+4t=R#)uGateway streaming consumer — bridges sync agent callbacks to async platform delivery. on_delta() queues deltas from the agent's worker thread; the async run() task buffers, rate-limits and progressively edits one platform message (send, then editMessageText; draft/native transports are optional per adapter). 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_CURSOR)is_intentional_silence_responseis_partial_silence_marker)ensure_closed_code_fences)StreamTransportMixin)StreamFallbackMixin)StreamThinkFilterMixinzgateway.stream_consumeru⏸ 等待审批中...c~]tRt^R/R?R@llt)RARBlt*RyRCRDllt+RERFlt,RGRHlt-RyRIRJllt.RKRLlt/RMRNlt0RORPlt1RQRRlt2RSRTlt3RURVlt4RWRXlt5RYRZlt6R[R\lt7R]R^lt8R_R`lt9RaRblt:RcRdlt;ReRfltRmRnlt?RoRplt@RqRrltARsRtltB]'RuRvl4tCRwtDR#)|GatewayStreamConsumeraAsync consumer that progressively edits a platform message with streamed tokens. Usage: ``agent.stream_delta_callback = consumer.on_delta``; ``create_task(consumer.run())``; after the agent finishes ``consumer.finish()`` then ``await task`` for the final edit.r_draft_id_counterNc8V^8dQhRRRRRRRRR R R R R RRR/#)r=adapterrchat_idrconfigzOptional[StreamConsumerConfig]metadatazOptional[dict]on_new_messagezOptional[callable]on_before_finalizezOptional[Callable[[], Any]]initial_reply_to_idr7run_still_currentzOptional[Callable[[], bool]]r")r?s"r.r@"GatewayStreamConsumer.__annotate__psdM,M,M,M,/ M, ! M, + M,8M,+M,8M,r-c  WnW nT;'g \4VnW@nWPnW`nWpn\\P!44Vn T;'gRVn RVn \VRR4RJVnRVn\"P$!4Vn\)4VnRVnRVnRVnRVn^VnVPP6Vn.Vn.VnRVnRVn RVn!VPE4RVn#RVn$^Vn%RVn&RVn'RVn(^Vn)\TVn+RVn,RVn-RVn.RVn/R#) cR#)Tr"r"r-r.0GatewayStreamConsumer.__init__..sr-TREQUIRES_EDIT_FINALIZEF@rr NApproval)0rLrMrcfgrO_on_new_message_on_before_finalize_initial_reply_to_idruuiduuid4_turn_id_run_still_currentstream_deltas_enabledgetattr_adapter_requires_finalize!_max_fallback_flood_retry_secondsqueueQueue_queueset_preview_message_ids _already_sent_edit_supported_last_edit_time_last_edit_overflowed_flood_strikesr_current_edit_interval_delivered_commentary_texts_delivered_segment_texts_in_think_block _think_buffer_before_finalize_notified_reset_message_state_use_draft_streaming _draft_id_draft_failures_egress_declined_use_native_streaming_native_stream_opened_native_last_pushed_len_DEFAULT_BOUNDARY_PLACEHOLDER_boundary_placeholder_boundary_reason_boundary_reopen_awaiting_reopen_after_boundary_reopen_seeded_eagerly) rDrLrMrNrOrPrQrRrSs &&&&&&&&&r.__init__GatewayStreamConsumer.__init__psx  3313  .#5 $7!DJJL) "3"E"E  &*"+2';SUZ*[_c*c'14.#(;;= /2e!"#"%*"&*hh&<&<#68(35%$).& !!# %*!(, !&%*"%*"'($&C" * %05,&+#r-cV^8dQhRR/#r=r>Noner")r?s"r.r@rTs''d'r-c RVnRVnR;VnVnRVnRVnRVnRVn\4Vn .Vn RVn VP4R#)zPPer-message (segment) state: fresh at construction and after each segment break.Nr F) _message_id_message_created_ts _accumulated_stream_ledger_last_sent_text_fallback_final_send_fallback_prefix#_fallback_preserve_partial_messagesrk_segment_preview_message_ids_tool_progress_lines_tool_progress_active_clear_turn_final_flagsrCs&r.rx*GatewayStreamConsumer._reset_message_statesj*.48 354D/!$)! "38069e)/1!+0" $$&r-cV^8dQhRR/#rr")r?s"r.r@rTs)))r-c LRVnRVnRVnRVnRVnR#)aReset every turn-final delivery flag to "nothing delivered yet". ``_delivered_final_text`` is the cleaned turn-final payload the gateway compares to the completed final_response before trusting the flags (a successful finalize edit may carry a stale preview); None = legacy trust. A payload-less ``_turn_split_delivery`` must NOT inherit legacy trust; ``_delivery_ambiguous`` (a full-final send timed out but MAY have landed) is the only case that does.FN)_final_response_sent_final_content_delivered_delivered_final_text_turn_split_delivery_delivery_ambiguousrCs&r.r-GatewayStreamConsumer._clear_turn_final_flagss.%*!(-%48"$)! $) r-cV^8dQhRR/#r<r")r?s"r.r@rTs  D r-c \\VP4RR4p\V4'g\VPRR4RJ#V!VP\ VP 44RJ# \ dR#i;i)zWhether THIS chat's transport treats the stream as the message: per-chat probe first (a relay adapter's class attribute only reflects its primary identity), else the legacy attribute; both on the CLASS (MagicMock-safe).stream_is_message_for_chatNdraft_stream_is_messageFT)retyperLcallablerrM Exception)rDprobes& r._stream_is_message(GatewayStreamConsumer._stream_is_messagestT\\*,H$O4<<)BEJdR R s4<<'89TA A  s 'A44 BBcV^8dQhRR/#r<r")r?s"r.r@rTs**t*r-c VP#)zJTrue only when native streaming is active (gates in-stream tool progress).)r}rCs&r.accepts_tool_progress+GatewayStreamConsumer.accepts_tool_progresss)))r-c V^8dQhRRRR/#)r=linerr>rr")r?s"r.r@rTs44S4T4r-c \V'd$VPP\V34R#R#)zTThread-safe: overlay a tool-progress line in the native bubble until the next delta.N)rjput_TOOL_PROGRESS)rDrs&&r.on_tool_progress&GatewayStreamConsumer.on_tool_progresss  KKOO^T2 3 r-cV^8dQhRR/#)r=r>rr")r?s"r.r@rTsOOOr-c RPVP4pRPRVPV344#)zFNative frame content: text, with any tool-progress lines below a rule. z --- c38"TFq'gK VxK R#5iNr").0ps& r. ?GatewayStreamConsumer._compose_frame_content..sN+HaA+Hs  )joinrr)rDprogresss& r._compose_frame_content,GatewayStreamConsumer._compose_frame_contents899T667ND,=,=x+HNNNr-finalF expect_editsc$V^8dQhRRRRRR/#)r=rrrr>z dict | Noner")r?s"r.r@rTs"  4 t Xc r-c VP'd\VP4M/pVP'dVPVR&V'dRVR&V'dRVR&T;'gR#)uPer-send metadata. ``final`` → notify=True (Mattermost treats notify-worthy sends as final when a broken thread root may fall back flat); ``expect_edits`` keeps editable previews on Telegram's legacy send path.reply_to_message_idTrnotifyN)rOdictr_)rDrrmetas&$$ r._metadata_for_send(GatewayStreamConsumer._metadata_for_sends^'+mmmtDMM"  $ $ $*.*C*CD& ' #'D !DN||tr-c VP#r)rmrCs&r.rWGatewayStreamConsumer.s ););r-c VP#r)rrCs&r.rWrs 0I0Ir-c VP#r)rrCs&r.rWrs t'7'7r-c VP#r)rrCs&r.rWrs D4Q4Qr-cV^8dQhRR/#rr")r?s"r.r@rTs ! !t !r-c R"VP'dR#RVnVPed\P!\4;_uu_4VP4p\ P !V4'd VGRjxL RRR4R#R#L +'giR#;i5i)z?Run the pre-finalize hook exactly once, swallowing hook errors.NT)rwr^ contextlibsuppressrinspect isawaitable)rDresults& r._notify_before_finalize-GatewayStreamConsumer._notify_before_finalizes|  ) ) ) )-&  # # /$$Y//113&&v.. LL0/ 0!0//s0A B'2BBB B'B B$  B'c V^8dQhRRRR/#r=textrr>rr")r?s"r.r@rTs$$$$r-c V'gR#VP'd"VPP4RVnV;PV, unV;PV, unR#)z=Append to the live buffer and the split-stable stream ledger.NF)rclearrrrrDrs&&r._append_accumulated)GatewayStreamConsumer._append_accumulatedsP   $ $ $  % % + + -).D & T! t#r-cV^8dQhRR/#rr")r?s"r.r@rT(s11t1r-c >VP;'g VPpVPP'dOVP VPP4'd$VR\ VPP4)pVP VR7R#)zMark the turn final as delivered by a prior mid-stream edit. Records what was ACKED on the wire, not ``_accumulated``: a throttled stream's last ack may be an older cursor-suffixed preview, which must not suppress the corrective send.Nrecord)rrr\rendswithlen_mark_final_delivered)rDackeds& r._mark_skip_redundant_finalize3GatewayStreamConsumer._mark_skip_redundant_finalize(sn$$99(9(9 88???u~~dhhoo>>1S112E ""%"0r-c V^8dQhRRRR/#)r=rr7r>rr")r?s"r.r@rT1s 4 4M 4T 4r-c PRVnRVnVeVPV4R#R#)zISet both turn-final flags; ``record`` also records the delivered payload.TN)rr_record_turn_final_payload)rDrs&&r.r+GatewayStreamConsumer._mark_final_delivered1s-$(!)-%    + +F 3 r-c V^8dQhRRRR/#r=rrr>r")r?s"r.r@rT>sVVSVSVr-c f\TPT;'gR44P4#)zONormalize like ``_send_or_edit`` output: directive strip + fence close + strip.r )r_clean_for_displaystriprs&&r._display_payload&GatewayStreamConsumer._display_payload>s'()@)@)LMSSUUr-c V^8dQhRRRR/#rr")r?s"r.r@rTBsAAsAtAr-c VP'dVP'd VPpVPV4VnR#)uRecord what the user actually saw as this turn's final answer. On a split ``text`` is only the trailing chunk, so the un-truncated ``_stream_ledger`` is recorded — else the gateway sees a mismatch and re-sends an answer the user already received.N)rrrrrs&&r.r0GatewayStreamConsumer._record_turn_final_payloadBs9  $ $ $)<)<)<&&D%)%:%:4%@"r-c V^8dQhRRRR/#)r= final_textrr>zOptional[bool]r")r?s"r.r@rTJs;;#;.;r-c jVPV4pV'gR#VPe7VPP4V8H;'gVPV4#VP'dR#VP 'dVPV4'dR#VP 'dR#R#)a~Tri-state reconcile of the recorded turn-final payload against ``final_text`` (a *successful* finalize edit can still carry a stale preview, so call success alone must not confirm delivery). True: recorded payload (or an earlier segment / commentary) matches. False: payload differs, or payload-less split. None: nothing recorded on a legacy/ambiguous path (caller trusts flags).NFT)rrrhas_delivered_textrrmr)rDrtargets&& r.delivered_final_matches-GatewayStreamConsumer.delivered_final_matchesJs &&z2  % % 1..446&@;;..z: <  $ $ $    $"9"9*"E"E///t:U:r-c V^8dQhRRRR/#r=rrr>rr")r?s"r.r@rThsMMsMtMr-c DaTPT;'gR4P4oVP4.VPOVPO5p\ S4;'d;\ ;QJdV3RlV4F 'gK R# R#!V3RlV44#)zDReturn True if *text* was already delivered as visible chat content.r c3H<"TFqP4S8HxK R#5irrrsentrs& r.r;GatewayStreamConsumer.has_delivered_text..m#LttJJLF$:t"TF)rr_visible_prefixrsrtranyrDrseenrs&& @r.r(GatewayStreamConsumer.has_delivered_texths((4::<$$&0)I)I0..0F|LL#Lt#LLL#Lt#L LLr-c V^8dQhRRRR/#rr")r?s"r.r@rTos M Ms Mt Mr-c aTPT;'gR4P4o.VPOVPOpVP'd VP VP 44\S4;'d;\;QJdV3RlV4F 'gK R# R#!V3RlV44#)ud``has_delivered_text`` restricted to deliveries that outlive the turn: commentary and finalized segments always count; the visible prefix only once ``_already_sent`` (a draft frame sets ``_last_sent_text`` but is ephemeral — a failed finalize send after it must still fall back to the gateway's real final send, same gate as ``delivered_final_matches``).r c3H<"TFqP4S8HxK R#5irrrs& r.rCGatewayStreamConsumer.has_durably_delivered_text..xr r TF) rrrsrtrmappendr rr rs&& @r.has_durably_delivered_text0GatewayStreamConsumer.has_durably_delivered_textos ((4::<R11RD4Q4QR     KK,,. /F|LL#Lt#LLL#Lt#L LLr-cV^8dQhRR/#rr")r?s"r.r@rTzs&&$&r-c DVPP\4R#)z>Finalize the current stream segment and start a fresh message.N)rjr _NEW_SEGMENTrCs&r.on_segment_break&GatewayStreamConsumer.on_segment_breakzs  %r-c(V^8dQhRRRRRRRR/#) r= placeholderz str | Nonereasonrreopenrr>zasyncio.Futurer")r?s"r.r@rT~s,//%/69/PT/ /r-c Rp\P!\4;_uu_4\P!4pRRR4V'dVP 4M\ PP4pVP'gVPR4V#T;'g\Vn T;'gRVn \V4VnRR/pVP P#\$WV34WV3# +'giL;i)aQueue an interaction boundary (approval / clarify prompt) from sync context. run() finalizes the current native stream (``placeholder`` when empty), then per ``reopen``: False (approval; unbounded waits) degrades to one send() at got_done; True (clarify) keeps native enabled so post-prompt output re-opens a fresh stream. Returns (Future, cancelled_flag); the Future resolves True once processed (cancelled_flag is legacy, no longer read). Without native streaming returns a bare, already-resolved Future.NTr[ cancelledF)rr RuntimeErrorasyncioget_running_loop create_future concurrentfuturesFuturer} set_resultrrrrrrjr_APPROVAL_BOUNDARY)rDrrr loopboundary_futurecancelled_flags&&&& r.close_for_approval_prompt/GatewayStreamConsumer.close_for_approval_prompt~s   . .++-D/26$,,.Jrr")r?s"r.r@rTs : :% :$ :r-c \P!4pVPP\V34TP \R\T44R7# \ dR#i;i)uBlock the agent worker thread until everything queued so far is delivered: ``(_FLUSH, Event)`` barrier — run() drains earlier items (FIFO), finalizes the segment, sets the event. False on timeout (consumer task may not be running).Fr)r7) threadingEventrjr_FLUSHrwaitmaxr)rDr7evts&& r.flush_pending_sync(GatewayStreamConsumer.flush_pending_syncs[oo  KKOOVSM *xxCw 8x99  s!A A,+A,cV^8dQhRR/#r<r")r?s"r.r@rTs44d4r-c tVP;'d&VP;'dVP'*#)zCNative stream, reopen requested after a boundary, nothing open yet.)r}rr~rCs&r._reopen_seed_pending*GatewayStreamConsumer._reopen_seed_pendings8**33t/S/S33222 4r-cV^8dQhRR/#rr")r?s"r.r@rTs**T*r-c tVP4'd"VPP\4R#R#)zThread-safe: request an EAGER native re-seed after a clarify answer. No-op unless reopen-pending, so a stray call can't open a spurious bubble mid-stream or on approval.N)rCrjr _REOPEN_SEEDrCs&r.request_reopen_seed)GatewayStreamConsumer.request_reopen_seeds(  $ $ & & KKOOL ) 'r-cV^8dQhRR/#rr")r?s"r.r@rTsIITIr-c VPeVP4R#R# \d\PRRR7R#i;i)z8Fire the on_new_message callback, swallowing any errors.Nzon_new_message callback errorTexc_info)r]rloggerdebugrCs&r._notify_new_message)GatewayStreamConsumer._notify_new_messagesG I##/$$&0 I LL84L H Is$"A  A cV^8dQhRR/#rr")r?s"r.r@rTs""d"r-c Ve>\P!\4;_uu_4VP4RRR4R#R# +'giR#;i)zWake a thread blocked in flush_pending_sync(), swallowing errors. Every loop path that consumed a ``_FLUSH`` barrier (incl. early ``continue``) must call this; a missed set stalls the caller for the full timeout.N)rrrrk)r4s&r. _signal_flush#GatewayStreamConsumer._signal_flushs<  "$$Y//!0/ #///s A A preserve_no_editc V^8dQhRRRR/#)r=rVrr>rr")r?s"r.r@rTs""""r-c lV'dVPR8XdR#VPVP4P4pV'dVPP V4VP 4VP'd+VP4'gVP4R#R#R#) __no_edit__N) rrrrrtrrxryr_bump_draft_id)rDrV finalizeds&$ r._reset_segment_state*GatewayStreamConsumer._reset_segment_states  0 0M A ++D,@,@AGGI   ) ) 0 0 ; !!#  $ $ $T-D-D-F-F    !.G $r-cV^8dQhRR/#rr")r?s"r.r@rTs666r-c ~\V4;P^, un\V4PVnR#)N)rrJrzrCs&r.rZ$GatewayStreamConsumer._bump_draft_ids( T $$)$d55r-cV^8dQhRR/#rr")r?s"r.r@rTs< KK 0//$$Y//o}==oFZFZF\F\#..{;0///sG>(D:D8D:A D:#+D:"G>0,E*E*. G>8D:: E'E"E>"E''E>* E; 5 G>>#G;!,G( G(  G;( G8 3G;;G>c V^8dQhRRRR/#)r=rlrr>rr")r?s"r.r@rTsstr-c "VP;'g VPpVPVRR7GRjxL 'd.\P RWP VP 4R#\PRWP 4\VPPVP V4GRjxL RR4'dR#\PR WP 4R#L \d!p\PRY4Rp?LRp?ii;iLe \d!p\PR Y4Rp?LzRp?ii;i5i) zClose the open native stream at a boundary; send() the pre-prompt text if that fails. False only when both finalize and the fallback send failed.TfinalizeNz0%s boundary: finalized stream (chat=%s, turn=%s)z %s boundary: finalize failed: %szY%s boundary: finalize not confirmed, falling back to send() for pre-prompt text (chat=%s)successFz*%s boundary: fallback send also failed: %sum%s boundary: both finalize and fallback send failed (chat=%s) — pre-prompt text may not have been delivered) rr _send_framerNrOrMrbrrhrerLsenderror)rDrl finalize_textrnsend_errs&& r.rd/GatewayStreamConsumer._finalize_boundary_streams,))GGT-G-G  K%%md%CCC O$llDMMCD N  . \T\\..t||]KKYX]^^_  Qll ,%D K NN=w J J KL \ NNG [ [ \s!EC:C8 C:+C:0#E.D*D(D*%E8C:: D%D E D%%E(D** E5E EEEc V^8dQhRRRR/#rr")r?s"r.r@rTs$$S$T$r-c |V'dVPPV4R#VfVP4R#R#)zThread-safe callback from the agent's worker thread. ``None`` signals a tool boundary: the current message is finalized and subsequent text goes out as a new message below any tool-progress messages.N)rjrrrs&&r.on_deltaGatewayStreamConsumer.on_deltas-  KKOOD ! \  ! ! #r-c V^8dQhRRRR/#)r=rr7r>rr")r?s"r.r@rTs$r-c Ve"VPP\V34VPP\4R#)zSignal stream completion. ``final_text`` is the AUTHORITATIVE completed final_response (incl. post-stream augmentation the accumulator never saw); the drain loop adopts it as the finalize payload. Interrupt/error paths call ``finish()`` bare.N)rjr _FINAL_TEXT_DONE)rDrs&&r.finishGatewayStreamConsumer.finishs/  ! KKOO[*5 6 r-cV^8dQhRR/#rr")r?s"r.r@rT sL'L'4L'r-c 8"VP4wVnVnVP4GRjxL VP 4'g,VP 4GRjxL VP 4R#VP4pVPe%VP!VP!GRjxL KVP'dVP4GRjxL KVP'dfVP4\VPVP 44'd,VP#4GRjxL VP 4R#VP%V4'EdVP 'g%VP&'dVP('dVP+4GRjxL VP&'gVP-4'dVP/V4GRjxL 'dVP 4R#VP-4'd=VP1VP24\4P6!R4GRjxL EK9VP9V4GRjxL VP'd-VP;V4GRjxL VP 4R#VP<e$VP?VP<4GRjxL VP@'dVPCV4GRjxL VPD'dVP1VP24\4P6!R4GRjxL EK:EL@ELELELEL7ELELELELLLLqL( \4PFdTPI4GRjxL M-\Jd!p\LPORT4Rp?MRp?ii;iTP 4R# TP 4i;i5i)z@Async task that drains the queue and edits the platform message.Ng?zStream consumer error: %s)(_resolve_length_budget_len_fn _safe_limit_start_transportsrc_abandon_native_stream_wake_flush_waiters _drain_queuer6ror5_eager_reopen_seedr1_flush_think_buffer _is_intentional_silence_responserr_suppress_silence_marker _should_editr}r_seal_overflow_heads_first_send_overflows_split_first_sendrTr4r$sleep _push_update_finalize_turnr8_deliver_commentaryr2 _end_segmentr3CancelledError _on_cancelledrrNrx)rDtickrns& r.runGatewayStreamConsumer.run sD)-)D)D)F& d&$$&&&H '..0055777F  $ $ &E((*))588$:P:PQQQ'''11333===,,.88O8OPTPaPa8bcc";;===`  $ $ &]$$T**%%%$*D*D*DIcIcIc 33555 555$:T:T:V:V!%!7!7!==="D  $ $ &C 5577 !..t/?/?@")--"555$++D111===--d333&  $ $ &#''32243G3GHHH)))++D111>>>&&t'7'78mmD)))E ' 8R4>6>624I1*%% '$$& & & 9 LL4a 8 8 9  $ $ &D $ $ &s1PM3PNN#M6$N)P;;N6M97NN"M<#N;A NM? NP N8N NN.NNNN.NN NP N64N*N+NN NN1N2N7P ,N5N6N N N!N74N+N,N3P6N9N<N?NNNN NNNNN(O0>O?O0PO0O0O+&P+O00P3PPPcV^8dQhRR/#)r=r>z tuple[Callable[[str], int], int]r")r?s"r.r@rTps[[(H[r-c VP4wrV\RVP4V!VPP4, ^d, 43#)zPer-chat length function (relay adapters differ per chat, e.g. utf16) + budget. isinstance gate: MagicMock auto-attributes aren't callables; test doubles use len.i)_fallback_len_budgetr=_raw_message_limitr\r)rDlen_fn_s& r.r,GatewayStreamConsumer._resolve_length_budgetpsG --/ s3 7 7 9F488??VP4\PRVPVP4R#R#L5i)uResolve native/draft transport; native wins (adapters declaring it can't edit). The empty seed frame shows "typing" before the first token; on failure → edit path.z7Stream consumer using native-stream transport (chat=%s)z4Native streaming seed frame raised; disabling nativeTrLNFzBStream consumer using native-draft transport (chat=%s draft_id=%s)) _resolve_native_streamingr}rNrOrM_try_seed_framer~ry_resolve_draft_streamingrZrzrCs&r.r'GatewayStreamConsumer._start_transportsys&*%C%C%E"  % % % LLRTXT`T` a))*`37*999-1*,1)).D &$($A$A$C!  $ $ $    ! LL]t~~ 7 %9sAC* C(! C*+=C*)AC*cV^8dQhRR/#)r=r>r0r")r?s"r.r@rTs&2&2e&2r-c \4pVPP4pT\ Jd RTnT#T\Jd RTnT#T\Jd RTn T#\T\4'dT'd T^,MRpT\JdTPT^,4KT\Jd@TP 'd,TP"P%T^,4RTnKKT\(JdT^,T^,3TnT#T\,JdT^,TnT#T\0JdR;TnTnT^,TnT#TP7T4EKd \PdTu#i;i)zDrain everything queued so far into one tick. Control sentinels stop the drain (they take effect this tick); _FINAL_TEXT / _TOOL_PROGRESS / text deltas fold into state so simultaneous items batch.TN)r0rj get_nowaitrhEmptyrr1rr2rGr5ritupler_adopt_final_textrr}rrrr+r6r3r8r;r3r4_filter_and_accumulate)rDritemkinds& r.r"GatewayStreamConsumer._drain_queues]w {{--/u} $  |#)-& |#'+$ (u55$47DD{"&&tAw/'-----44T!W=15D..++*.q'47);& $'+Aw$ :>>!7#'7  ++D1=;;   sE//F F c V^8dQhRRRR/#)r= final_rawrr>rr")r?s"r.r@rTs,,3,4,r-c VP'g'VP'gVP'gR#VP'gGVP V4pV'd,W P VP48wd WnWnR#VP pV'dbVP V4'dI\V4\V48d.V;PV\V4R, unWnR#R#R#R#)upAdopt the authoritative final (see finish()) as the finalize content — only if this consumer streamed something (a no-stream turn keeps the gateway's final-send ownership). Split delivery: wholesale adoption would repeat sealed heads, refusing makes the gateway resend the ENTIRE body — so append only the suffix when the final strictly prefix-extends the ledger.N)rrrrrr startswithr)rDr final_payloadledgers&& r.r'GatewayStreamConsumer._adopt_final_texts !!!T%5%5%59M9M9M ((( 33I>M2I2I$J[J[2\!\$-!&/# $$ i**622s9~F 7S   3v;rr")r?s"r.r@rTs88848r-c VP'gR#VPP'dR#VP'd+\ VP 4;'g VP pM\P!4VP, p\ W0P8;'dVP ;'gF\VP 4VPP8;'dVP'*4pT;'d*\VPVP 44'*#)z/Decide whether this tick flushes an edit/frame.TF)rEr\rr}rrrtime monotonicrorrrrrq_is_partial_silence_markerr)rDr should_editelapseds&& r.r"GatewayStreamConsumer._should_edits 88     % % %t001OOT5O5OKnn&)=)==G +F+F F \ \4K\K\ @ @#&t'8'8#9TXX=V=V#V$?$?+/+>+>'>AK 88#=  # #D$5$5 6$8 8 8r-c V^8dQhRRRR/#rr")r?s"r.r@rTs--E-d-r-c "VPVPVPVP4p\ V4^8:d2VP VPVPVP4pVP p\ V4^8pVRR F8pVPWSVPR7GRjxL pVeWc8XdRpMTpK: V'dBVR ,pR\ V4 R\ V4 R2pVPV4VnRVn RVn RVn RVn \P!4VnVP'dVP'*;'g&VP#VPRR 7GRjxL p T;'dT VnVP$'d$RVn VP'VPR 7R#VP('dRVnRVnR#ELKLq5i) zNo message to edit yet and the buffer overflows: seal only the head chunks; the tail stays in _accumulated as the active preview later deltas edit in place. True when the turn finished here (the run loop returns).N)rFz (/)Tr rsr)_truncate_for_streamrrrr_split_text_chunksr__send_new_chunkr1 removesuffixrrrrrrro _send_or_editrrr2rr) rDrchunksreply_toheads_deliveredchunknew_idtail indicatortail_delivereds && r.r'GatewayStreamConsumer._split_first_sends**4+<+N>NPTP\P\] v;! ,,T->->@P@PRVR^R^_F,,f+/CR[E//t}}/UUF~!3"'H ! ":DS[M3v;-q9I $ 1 1) >t>r-c NVPRJ;'dVP4#r)rrrCs&r.r+GatewayStreamConsumer._first_send_overflows%s"4'==DOO,==r-cV^8dQhRR/#rr")r?s"r.r@rT(s--D-r-c 8"VP4'dVPeVP'd\VPVP VP 4pVPPR^V4pW!^,8dTpVPRVpVPVRRR7GRjxL pVP'g V'gR#VPVRPR4VnRVnRVn RVn EKR#R#R#La5i)zVExisting message overflowing: seal it with the head, start a new message for the rest.NrTFrt is_turn_finalr ) rrrnrrrrrfindrrlstriprr)rD cp_budgetsplit_atroks& r.r*GatewayStreamConsumer._seal_overflow_heads(sooD$4$4$@TEYEYEY*4+<+N>NPTP\P\]I((..tQ BHq.($%%ix0E))%$e)TTB((( $ 1 1() < C CD ID #D #%D (,D %FZ$@Us)DDA=D6D7D DADc V^8dQhRRRR/#r=rr0r>rr")r?s"r.r@rT;s++u++r-c "VPpVP'dVP'dRVP4pV'd9VPP 'dW PP , pMW PP , pVP ;'d#VP;'dVPRJVn TPY!P ;'g VPVP R7GRjxL Vn \P!4VnRVnR#L,5i)zGSend/edit this tick's visible text (cursor-suffixed unless finalizing).NrF)rrEr}rr\rr1ryrr:rr2r9rrror)rDr display_texts&& r.r"GatewayStreamConsumer._push_update;s(( ???)))#::< DHHOOO HHOO3L/ (,}}(D(D9R9R(D(D+/+;+;t+C # %)$6$6 ==#J#JD4J4J--%7%)) $~~/%*" )s50EE E'A E3E1E9 EE-Ec V^8dQhRRRR/#rr")r?s"r.r@rTTs11141r-c v"VP'g!VPfVP'dVP4GRjxL VP 4'd5VP'g#\ P RVP4R#VP'drVP'd`VP'gNVP'g<VPR4GRjxL \ P RVP4R#VP'dVVP'g2TPVP;'gRRR7GRjxL R#VP4R#VP'dVPV4GRjxL R#R#ELhLLLL5i)zNgot_done: final edit without cursor, or one continuation send if edits failed.NudClarify reopen boundary with no post-prompt content — skipping lone-placeholder finalize (turn=%s)z$Eager-seed empty finalize failed: %suUEager reopen seed but no post-answer content — closed empty typing bubble (turn=%s)u✅Fr)rrrmrrCrNrOrbrr~r9_close_empty_native_bubbler}_finalize_editr_finalize_edit_pathrDrs&&r.r$GatewayStreamConsumer._finalize_turnTsV     0 0 <@R@R@R..0 0 0  $ $ & &t/@/@/@ LLLMQ]] \)))d.H.H.H'''0C0C0C112XY Y Y LL@AE P  ' ' '&&&))$*;*;*D*DuU)SSS**,    **40 0 0) 1 ZT 1s2F9F9F0 F9#F953F9)F9;F9 F9F93F347F9,F9>F9 F9(F5))F9F9'F7( F93F95F97F9c V^8dQhRRRR/#rr")r?s"r.r@rTns99e99r-c "VP'd&VPVP4GRjxL R#VP'dVP VPR7R#VP 'dIVP 'd%VP'gVP'dVP4R#VP'deVPVP4GRjxL 'g:VP'd&VPVP4GRjxL R#R#R#VP'g&VPVP4GRjxL R#R#EL@LLHL5i)zNEdit-transport finalize (the non-native got_done branches, in priority order).Nr) r_send_fallback_finalrrrr9rfrpr:rrrrmrs&&r.r)GatewayStreamConsumer._finalize_edit_pathns!  $ $ $++D,=,=> > >  & & &  & &d.?.? & @  $*I*I*I)-)C)C)CtGbGbGb  . . 0    ,,T->->???DD]D]D]//0A0ABBBE^?###%%d&7&78 8 8 $ ?@B 9s{1E<E3E< /E<<E<E< E<2#E<E<4E65 E<?E<E</E80E< E<*E:+ E<6E<8E<:E<rTc$V^8dQhRRRRRR/#)r=rrrrr>r")r?s"r.r@rTs!)))))r-c "VPVRR7GRjxL VnVP'dTPV'dTMRR7VP#LC5i)zRfinalize=True send_or_edit; on success mark the turn delivered (+ record payload).TrsNr)rrr)rDrrs&&$r.r$GatewayStreamConsumer._finalize_editsS*.*<*rr")r?s"r.r@rTs ( ( ( (r-c "VP4pV'gVP4VPV4GRjxL \P!4VnV'gVP4R#R#L:5i)uPost commentary as its own message. Cumulative transports keep the stream going — resetting _accumulated would break the append-only invariant / lose text.N)rr\_send_commentaryrrro)rDr8 cumulatives&& r.r)GatewayStreamConsumer._deliver_commentarys`//1   % % '##O444#~~/  % % ' 5s=A<A:%A<&A<c V^8dQhRRRR/#rr")r?s"r.r@rTs99u99r-c "VP4'dR#VP'd<VP'g*VPR8wdVP 4GRjxL VP RR7R#L5i)uTool boundary: edit-based transports reset so the next chunk is a fresh message. Cumulative transports must NOT reset — clearing _accumulated makes the next frame a non-prefix snapshot and the connector re-appends the whole answer. preserve_no_edit: "__no_edit__" (platform never returned a real id — Signal, github_comment webhook) must keep its sentinel or every tool boundary posts a new message; the continuation goes out once via _send_fallback_final.NrYT)rV)rrr9r#_flush_segment_tail_on_edit_failurer\rs&&r.r"GatewayStreamConsumer._end_segmentsh  % % ' '     d&9&9&9$$ 5::< < < !!4!8 =s*A>A>$A>#A<$A>cV^8dQhRR/#rr")r?s"r.r@rTsAATAr-c "RpVP'dnVP'd\\P!\4;_uu_4\ VP VPRRR7GRjxL 4pRRR4M&VPfVP4GRjxL V'd3VP'gVPVPR7R#R#R#Ls +'giLN;iLU5i)uBest-effort final edit on task cancel: finalize=True so REQUIRES_EDIT_FINALIZE platforms apply formatting; is_turn_final=False because this handler owns the flags. Only a successful edit confirms delivery — a partial send may be just "Let me search…", not the answer.FTrNr) rrrrrrrrrr)rDbest_effort_oks& r.r#GatewayStreamConsumer._on_cancelleds     !1!1!1$$Y//!%D,>,>%%E-?-K'K"L0/   %--/ / / $";";";  & &d.?.? & @#<> 'K0/ 0sL&C="C= 'C(2C& 3C(;*C=%C;& C=2C="C=&C(( C8 3 C=cV^8dQhRR/#rr")r?s"r.r@rTs00T0r-c T\P!\4;_uu_4VPP 4p\ V\ 4'gK3\V4^8XgKEV^,\JgKXVPV^,4Kr +'giR#;i)zwWake still-queued _FLUSH waiters so a consumer dying mid-flush doesn't stall flush_pending_sync() for its full timeout.N) rrrrjrrirrr;rT)rDrs& r.r)GatewayStreamConsumer._wake_flush_waiterssj  + +{{--/dE**s4yA~$q'VBS&&tAw/ , + +s/B B)B<B B' c V^8dQhRRRR/#rr")r?s"r.r@rTsMMMMr-c .\P!V4#)zRHide MEDIA: / [[audio_as_voice]] directives; media is delivered post-stream.)_BasePlatformAdapter"strip_media_directives_for_display)rs&r.r(GatewayStreamConsumer._clean_for_displays$FFtLLr-)7rrfrmrrwrrrrrrsrrtrr{rzrnr|rrrrrrqrur_rprorrrgrrrr~r^r]rlrjrrcrrrrvrrrbrryr}rLr\rMrOrd)NNNNNNr)Nr[F)rZ)Er#r$r%r&r'_MAX_FLOOD_STRIKESsecretsrandbitsrJr)rrxrrrGrrrr already_sentfinal_response_sent message_idfinal_content_deliveredrrrrrrrrrrr/r4r?rCrHrP staticmethodrTr\rZrordr~rrrrrrrrrrrrrrrrrrrrrrr,r"r-r.rIrIcs^ %--b1s1M,^'$)( **4 O % e ;2$L'`[7.&2P,&4"8.-^B>-&+21490)): (9"A"0 M  Mr-rI)>r' __future__rr$concurrent.futuresr'rrloggingrhrr9rr` dataclassesrtypingrrrgateway.platforms.baserrrgateway.configr r(r r*r r+gateway.response_filtersr rr rgateway.stream_consumer_fencesr!gateway.stream_consumer_transportr gateway.stream_consumer_fallbackrgateway.stream_consumer_thinkr getLoggerrNobjectrrr3rrr;r+rGr)r(rjrrr0rIr"r-r.r2s#  !**N5;;=EB@@   4 5 x h h Xx !3!3!:!:; !9  $ aa a&m M02EG]m Mr-