tjH="dZddlmZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z ddlmZmZddlmZddlmZmZdd lmZmZGd d ZejeZeGd d ZddddZGddZdS)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)AsyncProtocolTransportEventStreamHandlec6eZdZdZdZdddZdd ZddZdZdS) _SeenEventIdsz)LRU set of event ids with bounded memory.)_data_maxsize'maxsizeintreturnNonec:t|_||_dSN)rrr)selfrs ]/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/langgraph_sdk/stream/controller.py__init__z_SeenEventIds.__init__&s-8]]  event_idstrc||jvr|j|dSd|j|<t|j|jkr|jddSdS)NF)last)r move_to_endlenrpopitemrrs raddz_SeenEventIds.add*sr tz ! ! J " "8 , , , F# 8 tz??T] * * J  E  * * * * * + *robjectboolc||jvSr)rr&s r __contains__z_SeenEventIds.__contains__2s4:%%rc*t|jSr)iterrrs r__iter__z_SeenEventIds.__iter__5sDJrN)r)rrrr)rr rr)rr(rr)) __name__ __module__ __qualname____doc__ __slots__rr'r+r/rrrr!so33%I     ++++&&&&     rrcTeZdZUdZded<ded<eejZded<d S) _SubscriptionzDInternal record for one active subscription on a `StreamController`.ridr params)default_factoryz asyncio.QueuequeueN) r0r1r2r3__annotations__r asyncioQueuer;r5rrr7r7@sMNN GGG 5???E??????rr7g)delayhandlerr?floatrrcvK|rtj|d{V|d{VdS)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)r=sleepclose)r@r?s r _close_afterrEPsX  #mE""""""""" ,,..rceZdZdZdddddddd@dZddddAdZdBd ZdCd$ZdDd&ZeZ eZ dEd(Z dBd)Z e Z dBd*ZdFd,ZdGd.ZdHd0ZdHd1Z dIdJd5ZdKd8ZdLd;ZdKd<ZdMd=ZdNd?ZdS)OStreamControlleraManages 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). Nirg?g@)run_start_gatemax_queue_sizeseen_event_ids_maxmax_reconnect_attemptsreconnect_backoff_basereconnect_backoff_cap transportrrI$Callable[[], Awaitable[None]] | NonerJrrKrLrMrArNrrc||_||_t||_d|_i|_d|_d|_d|_t|_ d|_ d|_ ||_ ||_||_dS)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)rrOrIrJrKrLrMrNs rrzStreamController.__init__rs$-,5GHHH%&"8:8<<@"7;>Aee" #' '=$'=$&;###r) namespacesdepthchannels list[str]rclist[list[str]] | Nonerd int | NoneAsyncIterator[Event]cjdt|i}|||d<|||d<||S)aOpen 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. reNrcrd)list_subscription_iter)rrercrdr9s r subscribezStreamController.subscribesG$.tH~~">  !#-F<  #F7O&&v...rcK|jrdSd|_|jb|jtjt t j5|jd{Vdddn #1swxYwY|j|j d{V|j rt j |j ddid{VdSdS)z?Tear down the controller, awaiting any pending rotation closes.NTreturn_exceptions) r^r[cancel contextlibsuppress Exceptionr=CancelledErrorrYrDr]gatherr.s rrDzStreamController.closes9 <  F   (   $ $ & & &$Y0FGG ( ('''''''' ( ( ( ( ( ( ( ( ( ( ( ( ( ( (   *%++-- - - - - - - -  % V.$"<UPTUU U U U U U U U U U V VsA11A58A5r9r r7ct|j|tj|j}|xjdz c_||j|j<|S)zDAllocate a subscription id, create a bounded queue, add to registry.rR)r8r9r;rS)r7rWr=r>rUrXr8)rr9subs r_register_subscriptionz'StreamController._register_subscriptions\)-(<===    ""a'""&)CF# rsubscription_idc<|j|ddS)zARemove a subscription from the registry. No-op if already absent.N)rXpop)rrys r_unregister_subscriptionz)StreamController._unregister_subscriptions! 66666rAsyncGenerator[Event, None]cK||} |jr ||jdS||d{V| |jd{V}| ||jdS|WVD#||jwxYwr)rxr^r|r8_reconcile_stream_ensure_fanout_runningr;get)rr9rwitems rrlz#StreamController._subscription_iters))&11 2|   ) )#& 1 1 1 1 1((00 0 0 0 0 0 0 0  ' ' ) ) )  Y]]__,,,,,,<  ) )#& 1 1 1 1 1    ) )#& 1 1 1 1sB1AB1+B11C c|j|jr-tj||_dSdSr)r[doner= create_task_fanoutr.s rrz'StreamController._ensure_fanout_runningsH   $(9(>(>(@(@ $ ' 3DLLNN C CD    % $rcpKddlm}|jso|j}|dS ||j23d{V}|jrnVt |jD]-}|||j r|j |.e6n2#t$r%}td|Yd}~nd}~wwxYw|j|ur|jd{V}|t!|t"jsu|jsnt'jt5|jd{Vdddn #1swxYwY|d{V}|rnn|jo|jD]}|j ddS)a2Single 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.subscriptionrr^rY _dedup_itereventsrkrXvaluesr9r; put_nowaitrs_loggerdebugr isinstancer=rtrqrrrD_reconnect_shared_stream)rrsharedeventrwdrop_errerr reconnecteds rrzStreamController._fanouts KJJJJJ, (F~ H#'#3#3FM#B#B8888888%|#D$7$>$>$@$@AA88//szBB8I007778$C  H H H >++K88 !&&&&&&&&&)      ++G444444444 #-D 44us:A**4B! B!candidate_filtercKddlm}|j'|j ||jt |rdS||}|j||}|j}||_||_|j d{V|\tj t|}|j |||j jdSdS)aTEnsure the shared SSE covers `candidate_filter`. Rotate if not. Open-new-before-close-old: any events buffered server-side between the two opens are replayed on the new SSE, and `_seen_event_ids` dedupes the overlap. Awaits `new_stream.ready` so the HTTP connection is established before returning. r) filter_coversN)extra)rrrYrZdict_compute_current_unionrTrrrr=rrEr]r'add_done_callbackdiscard)rrr new_filterr old_streamtasks rrz"StreamController._reconcile_stream?s' DCCCCC   +*6 d8$?O:P:PQQ7 F007G0HH _66  # #J / /  ( (%/"  !&|J'?'?@@D  & * *4 0 0 0  " "4#=#E F F F F F " !rc<K||d{VS)z%Public alias for `_reconcile_stream`.N)r)rrs rreconcile_streamz!StreamController.reconcile_stream]s-++,<=========rrSubscribeParams | Nonedict[str, Any]cddlm}d|jD}|"|t |||S)Nr)compute_union_filterc6g|]}t|jSr5)rr9).0rws r z;StreamController._compute_current_union..fs/) ) ) !$D  ) ) ) r)rrrXrappendr)rrrfilterss rrz'StreamController._compute_current_unionasu KJJJJJ) ) (,(;(B(B(D(D) ) )    NN4;; ' ' '##G,,,rseqr c0||dS)zCAdvance the reconnect cursor from a command response meta sequence.N) _observe_seqrrs robserve_applied_through_seqz,StreamController.observe_applied_through_seqqs #rrr cV||ddS)Nr)rr)rrs r_observe_eventzStreamController._observe_eventus( %))E**+++++rcjt|tr|j ||jkr ||_dSdSdSr)rrr_rs rrzStreamController._observe_seqxsB c3   T\%9S4<=O=ODLLL  =O=OrcFt|}|j |j|d<|S)Nsince)rr_)rr9outs rrz#StreamController._filter_with_since|s%6ll < #>>> /3 - - - - - ,,,,rrG)r@rr?rArr) r3 __future__rr=rqloggingr collectionsrcollections.abcrrrr dataclassesr r typingr langchain_protocolr r langgraph_sdk.stream.transportrrr getLoggerr0rr7rErGr5rrrs  #""""" ######NNNNNNNNNNNN((((((((55555555TTTTTTTT        8 ' H % % @@@@@@@ @EH nnnnnnnnnnr