tjSddlmZddlZddlZddlmZmZddlmZddl m Z m Z m Z ddl mZedge fZ Gdd ZdS) ) annotationsN) AwaitableCallable)Any) ProtocolEventStreamTransformertransformer_requires_async) StreamChanneltuple[str, ...]ceZdZdZ d7dddddd8dZd9dZd:dZd;dZdd!Z d?d$Z d@d%Z dAd(Z d?d)Zd@d*ZdAd+ZdBd-Zdd.dCd2ZdDd6ZdS)E StreamMuxuuCentral event dispatcher for the streaming infrastructure. Owns the main event log and routes events through a transformer pipeline. StreamChannels with a name discovered in transformer projections are auto-wired so that every `push()` also injects a `ProtocolEvent` into the main log. StreamChannels without a name are local-only. Pass `is_async=True` when the mux will be consumed via async iteration (`handler.astream()`). All StreamChannel instances discovered during registration are automatically bound to the matching mode. Attributes: extensions: Merged projection dict across all registered transformers. Treat as read-only — mutations won't be reflected back in individual transformers' state. native_keys: Projection keys contributed by transformers with `_native = True`. NFT)is_async factoriesscope _assign_seq transformerslist[StreamTransformer] | Nonerboolrlist[TransformerFactory] | Nonerr rreturnNonec:||_||_||_t|_|j||j|g|_g|_d|_ d|_ i|_ t|_ i|_i|_|t!|nd|_d|_d|_g}g}||D]5}||} t)| ddr|n|| 6g||RD]} || |||pdD]*} t)| ddr|n|| +g||RD]} || dS)uyInitialize the mux and register transformers in order. Callers pass either `transformers` (pre-built instances) or `factories` (callables producing fresh instances per mux). Each transformer's `init()` is called, projections are merged into `extensions`, `_native` keys are recorded in `native_keys`, and any StreamChannel instances are bound and (if named) wired. Transformers with `StreamTransformer.before_builtins = True` are registered ahead of the rest, preserving relative order within each lane. This lets content-mutating transformers (PII redaction, content filters, etc.) run before built-ins like `MessagesTransformer` that eagerly snapshot text fields into their projections. See `StreamTransformer.before_builtins` for the contract and foot-guns. Args: transformers: Already-built transformer instances. Registered only on this mux — they are NOT cloned into child mini-muxes built by `_make_child`. Use `factories` for transformers that should propagate to nested scopes. is_async: True for async dispatch (`apush` / `aclose` / `afail`), False for the sync path. factories: One-argument callables `(scope) -> StreamTransformer`. Called once with this mux's `scope` here, and cloned again per child scope by `_make_child` so each sub-mux gets fresh instances. scope: The namespace the mux operates within. The root mux is `()`. _assign_seq: Internal flag for child muxes. Root muxes assign monotonic `seq` numbers when appending to their main event log; child muxes share forwarded event objects and must not mutate their envelopes. Raises: RuntimeError: If any transformer requires an async run but the mux is in sync mode. TypeError: If a transformer's `init()` doesn't return a dict. ValueError: If transformers' projection keys collide. rrNbefore_builtinsFr)rrrr _events_bind _bind_mux _transformers _channels_seq _push_seq extensionsset native_keys_projection_owners_transformer_by_keylist _factories_pump_fn _apump_fngetattrappend _registerclear) selfrrrrrprerestfactory transformers S/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/langgraph/stream/_mux.py__init__zStreamMux.__init__0sb! &+ &5B__  H--- t$$$6835 *,%(UU24AC  )4DOOO$ 48 ?C(*(*  $ & &%genn ";0A5IISCCt&%%%%,}t}} , , {++++ IIKKK JJLLL'-2  KK):EBB LSS T T    )S=4== ( (K NN; ' ' ' ' ( (keystrStreamTransformer | Nonec6|j|S)z@Return the transformer that contributed `key` to the projection.)r'get)r0r8s r5transformer_by_keyzStreamMux.transformer_by_keys'++C000r7intc0|xjdz c_|jS)N)r"r0s r5_next_push_seqzStreamMux._next_push_seqs !~r7fnCallable[[], bool]c||_||j_|jD] }||_ |jD] }t |dd}| ||!dS)aWire the sync pull callback onto every projection in this mux. Records the pump on the mux so child mini-muxes built by `_make_child` can inherit it. Propagates to: - the main event log (`self._events`) - every projection StreamChannel in `extensions` - any registered transformer that exposes `_bind_pump` (e.g. `MessagesTransformer` so `ChatModelStream` instances drive the shared pump from their cursors) _bind_pumpN)r*r _request_morer rr,)r0rCchr4binds r5 bind_pumpzStreamMux.bind_pumpsr %' ". " "B!B  -  K; d;;DR  r7Callable[[], Awaitable[bool]]c||_||j_|jD] }||_ |jD] }t |dd}| ||!dS)z!Async counterpart to `bind_pump`. _bind_apumpN)r+r_arequest_morer rr,)r0rCrHr4abinds r5 bind_apumpzStreamMux.bind_apumpsp&( #. # #B "B  -  KK==E b   r7c|jtdt|j|j|d}|j||j|j||j|S)aBuild a mini-mux with the same factories scoped to `scope`. Used by `SubgraphTransformer` to attach a fresh transformer pipeline to each discovered subgraph handle. The child mux inherits the current pump bindings (so cursors on its projection logs drive the root pump), carries the same factory list forward to any grandchild subgraphs, and does not assign `seq` numbers so forwarded events can be shared without mutating their envelope. Raises: RuntimeError: If the mux was not constructed with `factories=`. Mini-muxes require factories so each scope gets its own fresh transformer instances. NzStreamMux._make_child requires the mux to be constructed with `factories=`; pre-built transformers can't be cloned to a new scope.F)rrrr)r) RuntimeErrorr rr*rJr+rP)r0rchilds r5 _make_childzStreamMux._make_childs ? ")  o]     = $ OODM * * * > %   T^ , , , r7r4rct|r+js$tt|jd|}t |ts$tdt|jt|tj z}|rUd fdt|D}tdt|jd|tt|dd}j||| j |t|j}|D]}|j|<|j|<|r,j||d S) zRegister a single transformer. Calls `transformer.init()`, stores the transformer for event processing, binds any StreamChannel instances in the projection, and merges the projection into `extensions`. uz requires an async run — it overrides aprocess/afinalize/afail or sets requires_async=True. Use astream(), not stream().z1StreamTransformer.init() must return a dict, got z, c3>K|]}|dj|dVdS)z (owned by )N)r&).0r8r0s r5 z&StreamMux._register..sP%%DDT%.s]!!]33'q'*@AA r7) _collect_scheduled_tasksrgathernextr afinalizerxr ryrzr)r0pendingresults first_errr{r4r|rHs r5aclosezStreamMux.aclose|sI//11  #NGLtLLLLLLLLG$ I$,0 - $ $K $!++----------  $ $ $&"#K $.  B:     "  # "sA55 B ?BB cK|}|D]}||rtj|ddid{V|jD].} ||d{V#t $rY+wxYw|jD]}|js| ||j js|j |dSdS)a&Fail on the async lane. Cancels every scheduled task across all transformers, awaits them to completion, then runs each transformer's `afail` hook and auto-fails channels and the main event log. Args: err: The exception that ended the run. rTN) rcancelrrrafailrxr ryrr)r0r}rtaskr4rHs r5rzStreamMux.afails%//11  D KKMMMM  C.'BTBB B B B B B B B-  K !'',,,,,,,,,,     .  B:  |# # L  c " " " " " # #sA// A<;A<list[asyncio.Task[Any]]c$d|jDS)z@Return a snapshot of in-flight tasks scheduled via transformers.cbg|],}t|ddD]}||-S)_stream_scheduled_tasksr)r,done)rXr4rs r5 z6StreamMux._collect_scheduled_tasks..sZ    -FKK  99;;      r7)rrAs r5rz"StreamMux._collect_scheduled_taskss&  #1    r7r[rjdict[str, Any]r\cp|D]}t|tr|j|j||j7|r|jn d|j}d fd }| ||dS) aIBind and optionally wire StreamChannel instances in a projection. All StreamChannels are bound and tracked. Channels with a name are additionally wired for protocol auto-forwarding. Args: projection: The projection dict returned by a transformer's `init()`. native: True when the owning transformer is `_native`. Named channels owned by a native transformer use the channel name directly as the protocol method; user-defined channels are prefixed with `custom:`. rNzcustom: method_namer9rCallable[[Any], None]cdfd }|S)Nitemrrrc4|dSr)_forward)rrr0s r5rzAStreamMux._bind_and_wire.._make_forward.._forwards MM+t<<<<._make_forwards.======= (r7)rr9rr) valuesr`r rrrr r-name_wire)r0rjr\valuemethodrs` r5rfzStreamMux._bind_and_wires  &&(( 7 7E%// 7 T] 333%%%%%e,,,:)+1MUZZ7M7M7MF(((((( KK f 5 5666 7 7r7rrrcd|gttjdz|dd}|jr|xjdz c_|j|d<|j|dS)aInject a ProtocolEvent for a StreamChannel push. Forwarded events bypass the transformer pipeline to avoid infinite recursion (a transformer that pushes to a channel during `process()` would re-trigger itself). These events are visible in this mux's main event log but are not passed through transformers' `process()` methods. Only the root mux assigns `seq` to forwarded channel events. Args: method: The full protocol method (already with or without the `custom:` prefix; resolved by `_bind_and_wire`). item: The payload pushed onto the channel. roi) namespace timestampdata)r]rparamsr@rqN)r>timerr!rrs)r0rrros r5rzStreamMux._forwards  t!344     % IINII9E%L %     r7r) rrrrrrrr rrrr)r8r9rr:)rr>)rCrDrr)rCrKrr)rr rr )r4rrr)rorrr)rr)r}rxrr)rr)rjrr\rrr)rr9rrrr)r^ __module__ __qualname____doc__r6r=rBrJrPrTr.rsrzrrrrrrfrrr7r5r r s.8<d(59!# d(d(d(d(d(d(L1111(        D('('('('T%%%%:82%%%%:****X####6    =B777777@!!!!!!r7r ) __future__rrrcollections.abcrrtypingrlanggraph.stream._typesrrr langgraph.stream.stream_channelr TransformerFactoryr rr7r5rs"""""" //////// :99999013DDEq!q!q!q!q!q!q!q!q!q!r7