tj1ddlmZddlZddlmZddlmZmZmZm Z ddl m Z m Z m Z e rddlmZe dZGdd e eZdS) ) annotationsNdeque) AsyncIterator AwaitableCallableIterator) TYPE_CHECKINGGenericTypeVar) StreamMuxTceZdZdZd(ddd)d Zd*d Zd+dZd,dZd-dZd.dZ d/dZ d0dZ d0dZ d1dZ d1d Zd2d3d%Zd2d4d'ZdS)5 StreamChannelu.Single-consumer drainable queue for streaming events, with optional protocol auto-forwarding. When constructed with a `name`, the StreamMux auto-wires every `push()` to also inject a `ProtocolEvent` into the main event stream using the channel's name as the method. When constructed without a name, the channel is local-only — items are only visible to in-process consumers that iterate the channel directly. Items are popped off the front as the consumer advances — there is no retention beyond what's currently queued. A channel accepts exactly one subscriber; a second `__iter__` / `__aiter__` call raises. Use `tee(n)` / `atee(n)` for fan-out. Starts unbound — neither `__iter__` nor `__aiter__` is available until the StreamMux calls `_bind(is_async)`. After binding, only the matching iteration protocol works; the other raises `TypeError`. Pump wiring (set by the run stream, not by `_bind`): - `_request_more`: sync pump callable, returns True if a new event was produced. - `_arequest_more`: async pump coroutine factory, same contract. Memory is bounded by caller pace: both sync and async use caller- driven pumps, so each cursor advance produces at most one event. Lazy-subscribe: `push` appends to the local buffer only when a subscriber has registered. Auto-forward via `_wire_fn` always fires regardless of subscription state. Lifecycle (`close` / `fail`) is managed by the mux — transformers don't need to close their channels manually. N)maxlenname str | Noner int | NonereturnNonec||dkrtd||_t|_||_d|_d|_d|_d|_d|_ d|_ d|_ d|_ dS)aInitialize the channel. Args: name: Optional protocol channel name. When set, the StreamMux wires every `push()` to also inject a `ProtocolEvent` into the main event stream. Surfaced on the wire as `custom:` for user-defined transformers, or as `` for channels owned by a native transformer (`_native = True`). When `None`, the channel is local-only. maxlen: Accepted for forward compatibility; currently unused. The caller-driven pump bounds memory naturally for single-consumer use. Raises: ValueError: If `maxlen` is not a positive integer or `None`. Nrz3StreamChannel maxlen must be a positive int or NoneF) ValueErrorrr_items_maxlen_closed_error _is_async _subscribed _request_more_arequest_more_wire_fn_mux)selfrrs ]/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/langgraph/stream/stream_channel.py__init__zStreamChannel.__init__1sy$  &A++RSS S ,1GG #)  ,0 &* 8<DH48 &* muxr c||_dSN)r")r#r's r$ _bind_muxzStreamChannel._bind_muxYs  r&is_asyncboolc@|jtd||_dS)aSBind this channel to sync or async mode. Called by the StreamMux after transformer registration. Must be called exactly once before any iteration. Args: is_async: True to enable async iteration, False for sync. Raises: RuntimeError: If the channel has already been bound. NzStreamChannel is already bound)r RuntimeError)r#r+s r$_bindzStreamChannel._bind\s& > %?@@ @!r&fnCallable[[T], None]c||_dS)z8Install the auto-forward callback (called by StreamMux).N)r!)r#r0s r$_wirezStreamChannel._wireps  r&itemrc|jrT|jrtd|j|jnd}|j||f|j||dSdS)a$Append an item. Auto-forwards if wired. The local buffer append is a no-op when no subscriber is registered, but auto-forwarding always fires so wired events reach the main event log regardless of subscription state. Items are stored as `(stamp, item)` tuples where stamp is a monotonic counter from the owning mux. Stamps are stripped by the default cursors; raw stamped tuples are visible on `_items`. Raises: RuntimeError: If the channel is closed (and subscribed). z%Cannot push to a closed StreamChannelNr)rrr.r"_next_push_seqrappendr!)r#r4stamps r$pushzStreamChannel.pushxs   .| L"#JKKK26)2GDI,,...QE K  t} - - - = $ MM$      % $r&cd|_dS)zMark the channel as complete.TN)rr#s r$closezStreamChannel.closes  r&err BaseExceptionc"||_d|_dS)zqMark the channel as errored. Args: err: The exception to surface to the subscriber. TN)rr)r#r=s r$failzStreamChannel.fails   r& Iterator[T]c|jtd|jrtd|jrtdd|_|S)zSubscribe and return a sync cursor. Can be called only once. Raises: TypeError: If the channel is unbound or bound to async mode. RuntimeError: If the channel already has a subscriber. NVStreamChannel has not been bound yet. Register the transformer with a StreamMux first.uFThis StreamChannel is bound to async mode — use 'async for' instead.z@StreamChannel already has a subscriber; use .tee(n) for fan-out.T)r TypeErrorrr. _sync_cursorr;s r$__iter__zStreamChannel.__iter__s > !C  > X    R    """r&c#K |jr!|j\}}|VnE|jr|j|jdS|j%|s|js |jsdSndSnr))rpopleftrrrr#_stampr4s r$rEzStreamChannel._sync_cursors { #{2244   ;*+%#/))++;t| r&AsyncIterator[T]c|jtd|jstd|jrtdd|_|S)zSubscribe and return an async cursor. Can be called only once. Raises: TypeError: If the channel is unbound or bound to sync mode. RuntimeError: If the channel already has a subscriber. NrCu?This StreamChannel is bound to sync mode — use 'for' instead.zAStreamChannel already has a subscriber; use .atee(n) for fan-out.T)rrDrr. _async_cursorr;s r$ __aiter__zStreamChannel.__aiter__s > !C ~ Q    S  !!###r&cK |jr"|j\}}|WVnK|jr|j|jdS|j+|d{Vs|js |jsdSndSur))rrHrrr rIs r$rMzStreamChannel._async_cursors { #{2244   ;*+%$0!0022222222;t| r&ninttuple[Iterator[T], ...]c|dkrtd|dt|Ddgd fd tfd t|DS) a{Subscribe and return `n` independent sync iterators. Each branch has its own buffer; items pulled from the underlying cursor are copied into every branch. Branches are naturally bounded by caller pace since the sync pump is caller-driven. Args: n: Number of branches to create. Must be >= 1. Returns: A tuple of `n` iterators over the same underlying stream. Raises: TypeError: If the channel is unbound or bound to async mode. RuntimeError: If the channel already has a subscriber. ValueError: If `n` < 1. ztee() requires n >= 1c*g|]}tSr.0_s r$ z%StreamChannel.tee.. "="="=q577"="="=r&FirRrrAc3K|} |r|VnKdrdS t}n#t$r dd<YdSwxYwD]}||eNTr)rHnext StopIterationr7)r]bufr4bbuffers exhaustedsources r$branchz!StreamChannel.tee..branchs!*C ' '++--''''q\ 'F#F||('+ ! %'' 'sAAAc3.K|]}|VdSr)rWrYr]rgs r$ z$StreamChannel.tee..+111VVAYY111111r&)r]rRrrA)rrFrangetuple)r#rQrgrdrerfs @@@@r$teezStreamChannel.tees& q55455 5"="=E!HH"="="=G  ' ' ' ' ' ' ' ' 1111a111111r&tuple[AsyncIterator[T], ...]c(|dkrtd|dt|Ddgdgtjd fd t fd t|DS) aSubscribe and return `n` independent async iterators. Caller-driven fan-out: each branch's `__anext__` either pops from its own buffer or, under a shared `asyncio.Lock`, pulls one item from the underlying cursor and distributes it to every branch's buffer. Args: n: Number of branches to create. Must be >= 1. Returns: A tuple of `n` async iterators over the same underlying stream. Raises: TypeError: If the channel is unbound or bound to sync mode. RuntimeError: If the channel already has a subscriber. ValueError: If `n` < 1. rUzatee() requires n >= 1c*g|]}tSrWrrXs r$r[z&StreamChannel.atee..7r\r&FNr]rRrrKc,K|} |r|WVdrdddS4d{V|sdr dddd{V[ d{V}nS#t$rdd<Ydddd{Vt$r%}|d<dd<Yd}~dddd{Vd}~wwxYwD]}|| dddd{Vn#1d{VswxYwY r_)rH __anext__StopAsyncIteration Exceptionr7) r]rbr4ercrderrorrelockrfs r$rgz"StreamChannel.atee..branch<s!*C '++--''''Q<Qx+#AhF ' ' ' ' ' ' ' '!il!  ' ' ' ' ' ' ' ' ' ' ' ' '!%+%5%5%7%7777777-!!!'+ !   ' ' ' ' ' ' ' ' ' ' ' ' '%!!!#$a'+ !   ' ' ' ' ' ' ' ' ' ' ' ' '!%''' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' ' 'sH D(BDCD% C. C8DCD D D c3.K|]}|VdSr)rWris r$rjz%StreamChannel.atee..Urkr&)r]rRrrK)rrNrlasyncioLockrm)r#rQrgrdrwrerxrfs @@@@@@r$ateezStreamChannel.atee s( q55566 6!!"="=E!HH"="="=G -1F|~~ ' ' ' ' ' ' ' ' ' '21111a111111r&r))rrrrrr)r'r rr)r+r,rr)r0r1rr)r4rrr)rr)r=r>rr)rrA)rrK)rP)rQrRrrS)rQrRrro)__name__ __module__ __qualname____doc__r%r*r/r3r9r<r@rFrErNrMrnr|rWr&r$rrs@  D"+"+"+"+"+"+"+P""""(    ,####.($$$$.()2)2)2)2)2V52525252525252r&r) __future__rrz collectionsrcollections.abcrrrr typingr r r langgraph.stream._muxr rrrWr&r$rs""""""HHHHHHHHHHHH22222222220////// GCLLG2G2G2G2G2GAJG2G2G2G2G2r&