ޠjSSSKJr SSKrSSKrSSKJrJr SSKJr SSK J r J r J r SSK Jr \S/\ 4r"SS 5rg) ) annotationsN) AwaitableCallable)Any) ProtocolEventStreamTransformertransformer_requires_async) StreamChanneltuple[str, ...]c\rSrSrSrSSSSSS.SSjjjrSS jrSS jrSS jrSS jr S S jr S!Sjr S"Sjr S#Sjr S$SjrS"SjrS#SjrS$SjrS%SjrSS.S&SjjrS'SjrSrg)( StreamMuxu5Central 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_seqcX lX@lXPl[5UlURR US9 URR U5 /Ul/UlSUl SUl 0Ul [5Ul 0Ul0UlUb [!U5OSUlSUlSUl/n/nUbuUH0nU"U5n [)U SS5(aUOUR+U 5 M2 /UQUQ7Hn UR-U 5 M UR/5 UR/5 U=(d SH(n [)U SS5(aUOUR+U 5 M* /UQUQ7Hn UR-U 5 M g)uaInitialize 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) self transformersrrrrprerestfactory transformers O/var/www/html/gaurav/venv/lib/python3.13/site-packages/langgraph/stream/_mux.py__init__StreamMux.__init__0ssb! &+ &5B_  H- t$6835 *,%(U24AC  )4DO$ 48 ?C(*(*  $%en ";0A5IICt&% % -}t} {+ - IIK JJL'-2-KK):EBBS T T .)S=4=K NN; ')c8URRU5$)z@Return the transformer that contributed `key` to the projection.)r"get)r+keys r1transformer_by_keyStreamMux.transformer_by_keys''++C00r4cDU=RS- slUR$)N)r)r+s r1_next_push_seqStreamMux._next_push_seqs !~~r4cXlXRlURH nXlM URHn[ USS5nUcMU"U5 M g)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_morerrr')r+fnchr0binds r1 bind_pumpStreamMux.bind_pumpsR %' "..B! !--K; d;DR.r4cXlXRlURH nXlM URHn[ USS5nUcMU"U5 M g)z!Async counterpart to `bind_pump`. _bind_apumpN)r&r_arequest_morerrr')r+rArBr0abinds r1 bind_apumpStreamMux.bind_apumpsP&( #..B " !--KK=E b .r4cURc [S5e[URURUSS9nURbUR UR5 UR bURUR 5 U$)ajBuild 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. zStreamMux._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%rDr&rJ)r+rchilds r1 _make_childStreamMux._make_childs} ?? ")  oo]]   == $ OODMM * >> %   T^^ , r4c^[U5(a2TR(d![[U5RS35eUR 5n[ U[5(d![S[U5R35e[U5[TR5-nU(aHSRU4Sj[U555n[S[U5RSU35e[[USS55nTR R#U5 TR%X%S 9 TRR'U5 [U5RnUH!nUTR(U'UTR*U'M# U(a)TR,R'UR/55 UR1T5 g ) 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, c3N># UHnU<STRUS3v M g7f)z (owned by )N)r!).0r7r+s r1 &StreamMux._register..s1%,C'T%<%.s:$!!]3'q'*@*@AA$sAA A) _collect_scheduled_tasksrgathernextr afinalizerxrryrzr)r+pendingresults first_errr{r0r|rBs r1acloseStreamMux.aclose|s//1 #NNGLtLLG$ I$,0 --K $!++---. ..B::: !   "  #1M .  $&"#K' $sQ1C;C-C;"C5C6C:"C; 5C;C C8"C3' C;3C88C;c# UR5nUHnUR5 M U(a[R"USS06IShvN URHnUR U5IShvN M URH'nUR(aMURU5 M) URR(dURRU5 ggNNy![ a Mf=f7f)zFail 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) rcancelrrrafailrxrryrr)r+rrtaskr0rBs r1rStreamMux.afails//1D KKM ..'BTB B B--K !'',,,. ..B::: !||## LL  c "$ C-   sOA D C- D!C15C/6C1:"D AD/C11 C?;D>C??Dc URVVs/sH0n[USS5HnUR5(aMUPM M2 snn$s snnf)z@Return a snapshot of in-flight tasks scheduled via transformers._stream_scheduled_tasksr)rr'done)r+r0rs r1r"StreamMux._collect_scheduled_taskssO $11 1  -FK99; K 1   s (A  A rXc^UR5Hn[U[5(dMURTRS9 UR T5 TR RU5 URcMnU(a UROSUR3nSU4SjjnURU"U55 M g)aBind 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:c>^SUU4SjjnU$)Nc*>TRTU5 gr)_forward)item method_namer+s r1rAStreamMux._bind_and_wire.._make_forward.._forwards MM+t._make_forwards== (r4)rstrrzCallable[[Any], None]) valuesr]r rrrrr(name_wire)r+rhrYvaluemethodrs` r1rdStreamMux._bind_and_wires  &&(E%// T]] 3%%%e,::)+1UZZ 7MF( KK f 56)r4cSU/[[R"5S-5US.S.nUR(a$U=RS- slURUS'URR U5 g)a6Inject 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. rsi) namespace timestampdata)rZrparamsr;roN)inttimerrrrq)r+rrrss r1rStreamMux._forwardsf  t!34     IINI99E%L % r4)r&rrrr$r!r%rrr"rrrr rr) r,zlist[StreamTransformer] | Nonerrcrzlist[TransformerFactory] | Nonerr rrcrr)r7rrzStreamTransformer | None)rr)rAzCallable[[], bool]rr)rAzCallable[[], Awaitable[bool]]rr)rr rr )r0rrr)rsrrr)rr)rrxrr)rzlist[asyncio.Task[Any]])rhzdict[str, Any]rYrcrr)rrrrrr)r[ __module__ __qualname____firstlineno____doc__r2r8r<rDrJrOr)rqrzrrrrrrdr__static_attributes__rr4r1r r s.8<d(59!# d(4d( d( 3 d(  d(d( d(L1(  D('T%:82%:*X#6 =B7(7597 7@!r4r ) __future__rrrcollections.abcrrtypingrlanggraph.stream._typesrrr langgraph.stream.stream_channelr TransformerFactoryr rr4r1rsI" / :013DDEq!q!r4