iF<BdZddlmZddlmZdgZGddZdS)uStateful scrubber for reasoning/thinking blocks in streamed assistant text. ``run_agent._strip_think_blocks`` is regex-based and correct for a complete string, but when it runs *per-delta* in ``_fire_stream_delta`` it destroys the state that downstream consumers (CLI ``_stream_delta``, gateway ``GatewayStreamConsumer._filter_and_accumulate``) rely on. Concretely, when MiniMax-M2.7 streams delta1 = "" delta2 = "Let me check their config" delta3 = "" the per-delta regex erases delta1 entirely (case 2: unterminated-open at boundary matches ``^...``), so the downstream state machine never sees the open tag, treats delta2 as regular content, and leaks reasoning to the user. Consumers that don't run their own state machine (ACP, api_server, TTS) never had any defence at all — they just emitted whatever survived the upstream regex. This module centralises the tag-suppression state machine at the upstream layer so every stream_delta_callback sees text that has already had reasoning blocks removed. Partial tags at delta boundaries are held back until the next delta resolves them, and end-of-stream flushing surfaces any held-back prose that turned out not to be a real tag. Usage:: scrubber = StreamingThinkScrubber() for delta in stream: visible = scrubber.feed(delta) if visible: emit(visible) tail = scrubber.flush() # at end of stream if tail: emit(tail) The scrubber is re-entrant per agent instance. Call ``reset()`` at the top of each new turn so a hung block from an interrupted prior stream cannot taint the next turn's output. Tag variants handled (case-insensitive): ````, ````, ````, ````, ````. Block-boundary rule for opens: an opening tag is only treated as a reasoning-block opener when it appears at the start of the stream, after a newline (optionally followed by whitespace), or when only whitespace has been emitted on the current line. This prevents prose that *mentions* the tag name (e.g. ``"use tags here"``) from being incorrectly suppressed. Closed pairs (``X``) are always suppressed regardless of boundary; a closed pair is an intentional, bounded construct. ) annotations)TupleStreamingThinkScrubbercDeZdZUdZdZded<edeDZded<edeDZded<e d eezDZ d ed <d"dZ d"dZ d#dZ d$dZed%dZd&dZd'dZd(dZed)dZed#d Zd!S)*ranStateful scrubber for streaming reasoning/thinking blocks. State machine: - ``_in_block``: True while inside an opened block, waiting for a close tag. All text inside is discarded. - ``_buf``: held-back partial-tag tail. Emitted / discarded on the next ``feed()`` call or by ``flush()``. - ``_last_emitted_ended_newline``: True iff the most recent emission to the consumer ended with ``\n``, or nothing has been emitted yet (start-of-stream counts as a boundary). Used to decide whether an open tag at buffer position 0 is at a block boundary. )thinkthinking reasoningthoughtREASONING_SCRATCHPADTuple[str, ...]_OPEN_TAG_NAMESc#"K|] }d|dV dS)<>N.0names 9/home/piyush/.hermes/hermes-agent/agent/think_scrubber.py z StreamingThinkScrubber.Ys*'P'P D 'P'P'P'P'P'P _OPEN_TAGSc#"K|] }d|dV dS)Zs*(R(R$d(R(R(R(R(R(Rr _CLOSE_TAGSc#4K|]}t|VdS)N)len)rtags rrz StreamingThinkScrubber.]s(IICHHIIIIIIrint _MAX_TAG_LENreturnNonec0d|_d|_d|_dS)NFT _in_block_buf_last_emitted_ended_newlineselfs r__init__zStreamingThinkScrubber.__init___s$ 15(((rc0d|_d|_d|_dS)z4Reset all state. Call at the top of every new turn.Fr$TNr%r)s rresetzStreamingThinkScrubber.resetds +/(((rtextstrc*|sdS|j|z}d|_g}|re|jr~|||j\}}|dkrD|||j}|r || dnd|_d|S|||zd}d|_n||}|||\}} |u|dks |d|krc|\} } |d| } | rF|| } | r/| | | d|_ || d}-|dkrh|d|} | rF|| } | r/| | | d|_ d|_||| zd}|||j }|||j} t|| }|r|d| }|| d|_n |}d|_|rF||}|r/| || d|_ d|S|ed|S)zFeed one delta; return the scrubbed visible portion. May return an empty string when the entire delta is reasoning content or is being held back pending resolution of a partial tag at the boundary. r$NFr T)r'r&_find_first_tagr_max_partial_suffixjoin_find_earliest_closed_pair_find_open_at_boundary_strip_orphan_close_tagsappendendswithr(rmax)r*r.bufout close_idx close_lenheldpairopen_idxopen_len start_idxend_idx preceding held_close emit_texts rfeedzStreamingThinkScrubber.feedjs 2i$ Q $~P $'+';';)(($ 9?? 33C9IJJD/3 ;TEFF DI773<<')i/001!&66s;;&*%@%@&&"( #NNd1g&9&9)-&Iw #JYJI $($A$A)$L$L $JJy111 ) 2 24 8 8!<ghh-Cr>>!$IXII $($A$A)$L$L $JJy111 ) 2 24 8 8!<&*DNh1223C //T_EE!55) 4,,# #FdUF I #TEFF DII #I "DI $ = =i H HI  9---%..t448wws||#cQ $fwws||rc|jrd|_d|_dS|j}d|_|sdS||}|r|d|_|S)u/End-of-stream flush. If still inside an unterminated block, held-back content is discarded — leaking partial reasoning is worse than a truncated answer. Otherwise the held-back partial-tag tail is emitted verbatim (it turned out not to be a real tag prefix). r$Fr2)r&r'r8r:r()r*tails rflushzStreamingThinkScrubber.flushsq > DI"DN2y  2,,T22  C/3}}T/B/BD , rr<tagsTuple[int, int]c|}d}d}|D]L}||}|dkr|dks||kr|}t|}M||fS)zfReturn (earliest_index, tag_length) over *tags*, or (-1, 0). Case-insensitive match. r1r)lowerfindr)r<rM buf_lowerbest_idxbest_lenridxs rr3z&StreamingThinkScrubber._find_first_tagsxIIKK  $ $C..--Cbyyh"nnhs88!!rc|}d}t|j|jD]\}}|}|}||}|dkrI|||t |z} | dkrv| t |z} | ||dkr|| f}|S)aReturn (start_idx, end_idx) of the earliest closed pair, else None. A closed pair is ``...`` of any variant. Matches are case-insensitive and non-greedy (the closest close tag after an open tag wins), matching the regex ``.*?`` semantics of ``_strip_think_blocks`` case 1. When two tag variants could both match, the one whose open tag appears earlier wins. Nr1r)rPziprrrQr) r*r<rRbestopen_tag close_tag open_lower close_lowerrBr>rEs rr6z1StreamingThinkScrubber._find_earliest_closed_pairsIIKK )-#&t8H#I#I + + Hi!))J#//++K ~~j11H2~~!XJ7IB#k"2"22G|x$q'11 '* ralready_emitted list[str]c,|}d}d}|jD]q}|}d} |||} | dkrn;||| |r|dks| |kr| }t |}n| dz}Xr||fS)zReturn the earliest block-boundary open-tag (idx, len). Returns (-1, 0) if no boundary-legal opener is present. r1rT)rPrrQ_is_block_boundaryr) r*r<r]rRrSrTr tag_lower search_startrUs rr7z-StreamingThinkScrubber._find_open_at_boundarysIIKK ? ' 'C IL 'nnY =="99**3_EE2~~x#&#&s88"Qw  '!!rrUboolcd|dkr$|r|ddS|jS|d|}|d}|dkr?|r|dd}n|j}|o|dkS||dzddkS)aTrue iff position *idx* in *buf* is a block boundary. A block boundary is: - buf position 0 AND the most recent emission ended with a newline (or nothing has been emitted yet) - any position whose preceding text on the current line (since the last newline in buf) is whitespace-only, AND if there is no newline in the preceding buf portion, the most recent prior emission ended with a newline rr1r2Nr$r`)r:r(rfindstrip)r*r<rUr]rFlast_nl prior_newlines rraz)StreamingThinkScrubber._is_block_boundary*s !88 :&r*33D9993 3I //$'' b== A / 3 < >A%%)*>*>v*F*F%HHHHH qrc.d|vr|S|}g}d}|t|krd}|||dzdkr|jD]}|}t|}||||z|krJ||z} | t|kr,|| dvr"| dz } | t|kr || dv"| }d}n|s ||||dz }|t|kd|S) aRemove any close tags from *text* (orphan-close handling). An orphan close tag has no matching open in the current scrubber state; it's always noise, stripped with any trailing whitespace so the surrounding prose flows naturally. rrFz r`Tr$)rPrrr9r5) rnr. text_lowerr=rpmatchedrrbtag_lenjs rr8z/StreamingThinkScrubber._strip_orphan_close_tagscsH t  KZZ\\  #d))mmG!AE'"d**?  C # I!)nnG!!AK-0I==K#d))mmQ90D0DFA #d))mmQ90D0D"&>  47###Q##d))mm$wws||rN)r!r")r.r/r!r/)r!r/)r<r/rMr r!rN)r<r/)r<r/r]r^r!rN)r<r/rUrr]r^r!rd)r<r/rMr r!r)__name__ __module__ __qualname____doc__r __annotations__tuplerrr;r r+r-rIrL staticmethodr3r6r7ra classmethodr4r8rrrrr@s  (O#(%'P'P'P'P'P"P"PJPPPP#(5(R(R/(R(R(R#R#RKRRRRII [0HIIIIILIIII6666 0000 ````D."""\""8""""2!5!5!5!5F[*[rN)r{ __future__rtypingr__all__rrrrrsz66p#""""" # $BBBBBBBBBBr