ޠjH=SrSSKJr SSKrSSKrSSKrSSKrSSKJr SSK J r J r J r J r SSKJrJr SSKJr SSKJrJr SS KJrJr "S S 5r\R4"\5r\"S S 55rSS.SSjjr"SS5rg)aStream controller: subscription registry and fan-out for AsyncThreadStream. `StreamController` manages the set of active subscriptions against one shared SSE connection, routing events from the shared stream to per-subscription queues. It is the centralised place for: - subscription registration / teardown - shared-stream lifecycle (open, rotate, close) - dedup of replayed events across rotations - fan-out from the shared stream to subscriber queues ) annotationsN) OrderedDict)AsyncGenerator AsyncIterator AwaitableCallable) dataclassfield)Any)EventSubscribeParams)AsyncProtocolTransportEventStreamHandlecD\rSrSrSrSrS S SjjrS SjrS SjrSr Sr g) _SeenEventIds!z)LRU set of event ids with bounded memory.)_data_maxsizec.[5UlXlgN)rrr)selfmaxsizes Y/var/www/html/gaurav/venv/lib/python3.13/site-packages/langgraph_sdk/stream/controller.py__init___SeenEventIds.__init__&s-8]  cXR;aURRU5 gSURU'[UR5UR:aURR SS9 gg)NF)last)r move_to_endlenrpopitemrevent_ids radd_SeenEventIds.add*s] zz ! JJ " "8 , # 8 tzz?T]] * JJ  E  * +rcXR;$r)rr"s r __contains___SeenEventIds.__contains__2s::%%rc,[UR5$r)iterrrs r__iter___SeenEventIds.__iter__5sDJJrN)')rintreturnNone)r#strr0r1)r#objectr0bool) __name__ __module__ __qualname____firstlineno____doc__ __slots__rr$r'r,__static_attributes__rrrr!s3%I +& rrcX\rSrSr%SrS\S'S\S'\"\RS9r S\S 'S r g ) _Subscription@zDInternal record for one active subscription on a `StreamController`.r/idr params)default_factoryz asyncio.Queuequeuer<N) r5r6r7r8r9__annotations__r asyncioQueuerCr;r<rrr>r>@s#N G  ?E=?rr>g)delayc# U(a[R"U5IShvN UR5IShvN gNN7f)zClose a handle, optionally after a brief delay. Used to detach closing the old stream from the synchronous rotation step so the new stream can absorb server-side replayed events first. N)rEsleepclose)handlerGs r _close_afterrLPs3  mmE""" ,,. #s!AAAAAAc4\rSrSrSrSSSSSSS .SS jjrSSS .SS jjrS S jrS!SjrS"Sjr \r \ r S#Sjr S Sjr \ rS SjrS$SjrS%SjrS&SjrS&SjrS'S(SjjrS)SjrS*SjrS)SjrS+SjrS,SjrSrg)-StreamController`aManages subscriptions and fan-out against one shared SSE connection. Responsibilities: - subscription registry (register / unregister) - shared-stream lifecycle (open on first subscribe, rotate on filter widen) - dedup of replayed events via a bounded LRU `_SeenEventIds` - fan-out from the shared stream to per-subscription queues Args: transport: the `AsyncProtocolTransport` bound to this thread session. run_start_gate: zero-argument async callable that resolves once the current `run.start` has committed server-side (no-op when no run is in flight). max_queue_size: per-subscription queue bound (default 1024). seen_event_ids_max: LRU cap for the dedup set (default 10_000). Nir.g?g@)run_start_gatemax_queue_sizeseen_event_ids_maxmax_reconnect_attemptsreconnect_backoff_basereconnect_backoff_capcXlX0l[US9UlSUl0UlSUlSUlSUl[5Ul SUl SUl XPl X`lXplg)NrF) _transport_max_queue_sizer_seen_event_ids_next_subscription_id_subscriptions_shared_stream_shared_stream_filter _fanout_taskset_rotation_close_tasks_closed_cursor_max_reconnect_attempts_reconnect_backoff_base_reconnect_backoff_cap)r transportrQrRrSrTrUrVs rrStreamController.__init__rso$-,5GH%&"8:8<<@"7;>Ae" #' '=$'=$&;#r) namespacesdepthcZS[U50nUbX$S'UbX4S'URU5$)zOpen a typed subscription against the shared SSE. Returns an async iterator that yields raw `Event` dicts matching the given filter. Multiple concurrent subscribes share one HTTP connection whose union expands or rotates as subscriptions come and go. channelsrkrl)list_subscription_iter)rrnrkrlrAs r subscribeStreamController.subscribes>$.tH~">  !#-<  #7O&&v..rc# UR(agSUlURb`URR5 [R"[ [ R5 URIShvN SSS5 URb"URR5IShvN UR(a)[ R"URSS06IShvN ggNv!,(df  Nz=fNRN7f)z?Tear down the controller, awaiting any pending rotation closes.NTreturn_exceptions) rdracancel contextlibsuppress ExceptionrECancelledErrorr_rJrcgatherr+s rrJStreamController.closes <<     (    $ $ &$$Y0F0FG''''H    *%%++- - -  % %..$"<"<UPTU U U &(HG . UsNA*D ,C5<C3=C52D 3D48D ,D-D 3C55 D?D D c[URU[R"URS9S9nU=RS- slX R UR 'U$)zDAllocate a subscription id, create a bounded queue, add to registry.rX)r@rArCrY)r>r]rErFr[r^r@)rrAsubs r_register_subscription'StreamController._register_subscriptionsT))--(<(<=  ""a'"&)CFF# rc<URRUS5 g)zARemove a subscription from the registry. No-op if already absent.N)r^pop)rsubscription_ids r_unregister_subscription)StreamController._unregister_subscriptions 6rc# URU5nUR(aURUR5 gUR U5IShvN UR 5 UR R5IShvN nUcURUR5 gU7v MIN^N-!URUR5 f=f7fr)r~rdrr@_reconcile_stream_ensure_fanout_runningrCget)rrAr}items rrp#StreamController._subscription_iters))&1 2||  ) )#&& 1((0 0 0  ' ' ) YY]]_,<  ) )#&& 1  1-  ) )#&& 1sKCB:CB:B62B: B8 B:C/B:8B::CCcURbURR5(a*[R"UR 55Ulggr)radonerE create_task_fanoutr+s rr'StreamController._ensure_fanout_runningsA    $(9(9(>(>(@(@ ' 3 3DLLN CD )Arc@# SSKJn UR(GdzURnUcgUR UR 5ShvN nUR(a O`[ URR55H7nU"X4R5(dMURRU5 M9 M|URULaURIShvN nUb[!U["R$5(dxUR(dg[&R("[5 URR+5IShvN SSS5 UR-5IShvN nU(aGMwOUR(dGMzURR5HnURRS5 M gGN GN![a!n[RSU5 SnAGN;SnAff=fGN%N!,(df  N=fN7f)aSingle consumer of the shared SSE; routes events to subscriptions. Why: rotation in `_reconcile_stream` replaces `_shared_stream` mid-loop. Re-read `self._shared_stream` on each outer iteration so we always consume from the current handle. The old handle's iterator exhausts naturally after `_close_after` closes it. On a post-ready transport drop (non-cancelled error in `shared.done`), attempts to reconnect up to `_max_reconnect_attempts` times before giving up and closing subscriber queues. r)matches_subscriptionNztransport drop in fanout: %r)!langgraph_sdk.stream.subscriptionrrdr_ _dedup_itereventsror^valuesrArC put_nowaitrx_loggerdebugr isinstancerEryrvrwrJ_reconnect_shared_stream)rrsharedeventr}drop_errerr reconnecteds rrStreamController._fanouts K,,,((F~ H#'#3#3FMM#B8%||#D$7$7$>$>$@A/zzBBII007 B ""f,"KK'O&sG,B,BCC LL#,,Y7"11779998(,(E(E(G"GK" 5,,,<&&--/C II  &038#B  H 8H>HH H  HHc# URnURn[X2SU--5n[R"SUS-5n[ R "XE-5IShvN gN7f)zJSleep with exponential backoff and jitter for reconnect attempt *attempt*.rg?N)rgrhminrandomuniformrErI)rattemptbasecaprGjitters r_reconnect_sleep!StreamController._reconnect_sleepsV++))CG,-54<0mmEN+++sAA)!A'"A)c# URnUcg[UR5H[nUR(a gURR UR U55nURIShvN X0l g gN![Ra e[a URU5IShvN Mf=f7f)zAttempt to reopen the shared stream after a transport drop. Returns True if a new stream was successfully opened, False if all reconnect attempts were exhausted or the controller was closed. NFT) r`rangerfrdrZopen_event_stream_filter_with_sincereadyrEryrxrr_)r base_filterr new_streams rr)StreamController._reconnect_shared_streams00  T99:G|| !__>>++K8 !&&&& #- ;'))  ++G444 sFP rcT[U5nURbURUS'U$)Nsince)rre)rrAouts rr#StreamController._filter_with_since|s'6l << #<)rr/r0r1)rAr r0zAsyncGenerator[Event, None])rr/r0r1)r0r4)rr r0r1r)rzSubscribeParams | Noner0dict[str, Any])rr r0r1)rr r0r1)rArr0r)rrr0r)r5r6r7r8r9rrqrJr~rregister_subscriptionunregister_subscriptionrprensure_fanout_runningrrrrrrrrrrrr;r<rrrNrN`s6*@D""(&'(+'*<*<= <  <  <!$<!&< %<  /3 -+ -  - ,rrN)rKrrGrr0r1) r9 __future__rrErvloggingr collectionsrcollections.abcrrrr dataclassesr r typingr langchain_protocolr r langgraph_sdk.stream.transportrrr getLoggerr5rr>rLrNr<rrrs # #NN(5T  8   H % @@ @EH nnr