"jGSrSSKJr SSKrSSKrSSKrSSKJrJrJ r J r J r SSK J r Jr SSKJrJr SSKJr SSKJr SS KJr SSKrSSKrSS KJrJrJrJr S S KJ r \"S 5r!\"S5r"\"S5r#\$\%\&S-\&S-\S-4r'SSjr(SSjr)SSjr*SSjr+SSjr,SSjr-S Sjr.S!Sjr/"SS\\"5r0g)"aBridge async streaming context managers to synchronous code on the caller's event loop. The synchronous streaming wrappers (`Agent.run_stream_sync` and `direct.model_request_stream_sync`) need to drive an async stream from sync code. Pumping via repeated `loop.run_until_complete(anext(...))` runs each step in a *different* asyncio task, so any cancel scope the async code enters and exits per step (e.g. the agent graph's per-node scopes, or `group_by_temporal`'s debouncer) straddles tasks and raises `RuntimeError: Attempted to exit cancel scope in a different task than it was entered in`. It also leaves OpenTelemetry spans dangling, since the run span never closes in the task that opened it. `SyncStreamBridge` instead keeps a long-lived task holding the async context manager open, and each streaming pass runs its entire `async for` in another long-lived task. All tasks run on the caller's event loop, preserving the event-loop affinity of async clients and other resources reused across sync calls. ) annotationsN)AsyncGenerator AsyncIterator AwaitableCallableIterator)AbstractAsyncContextManagersuppress)Context copy_context) get_ident) TracebackType)Generic)TypeIsTypeVar TypeVarTupleUnpack)_utilsTStreamT _PosArgsTc.[R"U5$)zMNarrow an optionally awaitable result without losing its generic return type.)inspect isawaitable)values U/home/mande/repo/quber/.venv/lib/python3.13/site-packages/pydantic_ai/_sync_stream.py _is_awaitabler(s   u %%c# UR5IShvN nUR U[ 545 UIShvN nUR"U6IShvN gNB![[4a e[anUR U5 SnAgSnAff=fNVNA![a>nUR[U5XDR5IShvN (deSnAgSnAff=f7f)zZEnter and exit `cm` in one task, remaining parked while sync code uses the yielded stream.N) __aenter__KeyboardInterrupt SystemExit BaseException set_exception set_resultr __aexit__type __traceback__)cmenteredexit_requestedstreamexc exit_infos r_hold_context_managerr0-s }}&  /0'(( llI&&&+' z *  c") ' \\$s)S2C2CDDD EsC#AAAC#BBBC#BC#AB6B C# BC#BC# C "(C C  CC#C  C#c# UR5(d[R"U45IShvN [R"S5IShvN gN#N7f)z`Wait for a task, then yield once so queued loop-stop callbacks run before this waiter completes.Nr)doneasynciowaitsleep)tasks r_wait_for_taskr7Ls> 99;;llD7### --  $s!0AAAAAAc|UR[U55nUR5(d)URU5 UR5(dM)UR5 g![aC UR 5(d*UR 5(dUR5(aeNtf=f![av [[5 URU5 SSS5 O!,(df  O=f[[5 UR5 SSS5 e!,(df  e=ff=f)zODrive a task to completion despite stale `run_until_complete()` stop callbacks.N) create_taskr7r2run_until_complete RuntimeError is_closed is_runningr$r exceptionresult)loopr6waiters r_run_task_to_completionrBSs   nT2 3F++-- ''/++--  KKM  >>##t'8'8FKKMM=J  m $  # #F +% $ $ m $ NN % % $ s_B;A+B;+A B85B;7B88B;;D;C/& D;/ C= 9D;D) D;) D7 3D;cJ[U5nUHnUR5 M UH&n[[5 [ X5 SSS5 M( UR 5 UR 5(dURU5 [ X5 g!,(df  M{=f)z]Tell the owner task to exit the stream context manager, then drive its cleanup to completion.N)tuplecancelr r$rBclearr2r&)r@ owner_taskr, pump_tasksr/tasksr6s r_shutdown_looprJjs * E  m $ #D /% $    !!),D- % $s B B" cx# [U5nUHnUR5 M U(a[R"USS06IShvN UR 5 UR 5(dUR S5 [[5 UIShvN SSS5 gN]N!,(df  g=f7f)zMCancel active stream pumps before allowing the context-manager owner to exit.return_exceptionsTNNNN) rDrEr3gatherrFr2r&r r$)rGr,rHrIr6s r _request_exitrOs * E  nne>TR[TTT55 gN)r9rO)r,r@rGrHsr request_exit$_finalize_loop..request_exits z>:NOrrMreturnNone) r<r2r r=r3get_running_loopr;r r$rJcall_soon_threadsafe)r@rGr,rHowner_thread_idrSs```` r_finalize_loopr[s ~~:??,,PP{o%doo.?.?   $ $ & ,  !!,/   -(tZM_`)((   s6BC'C$7C C$ C C$#C$' C5c# UR5IShvN $N![Ra [Rs$f=f7f)zQReceive one item without leaking `EndOfStream` through an asyncio task traceback.N)receiveanyio EndOfStreamrUNSET)receive_streams r _receive_onerbs=#++----   ||s1AA$AAAAc\rSrSr%SrS\S'SSjrSSjrSSjrSSSjjr SS jr SS jr SS jr \ SS j5rSS jrSrg)SyncStreamBridgeaRuns an async streaming context manager on the caller's event loop and bridges it to sync. Constructing the bridge enters `cm` in a long-lived owner task and exposes the yielded object as [`stream`][pydantic_ai._sync_stream.SyncStreamBridge.stream]. Cancel scopes entered and exited by the async code never straddle tasks, OpenTelemetry spans stay correctly nested, and async resources remain on the same event loop across synchronous calls. The owning sync wrapper calls [`shutdown`][pydantic_ai._sync_stream.SyncStreamBridge.shutdown] (from its own `__exit__`) to exit the stream. A `weakref.finalize` fallback requests the same cleanup if the wrapper is dropped without being closed, but callers should use the wrapper as a context manager for deterministic cleanup. rr-c [R"5 [SUS35e![a Of=f[R"5n[ 5nUR 5nUR 5nUR[XU55nURU5upO![a UR5(dUR5 [[5 [X75 SSS5 O!,(df  O=f[[5 UR5 SSS5 e!,(df  e=ff=fXlX0lXplX`lX@lXl[-5Ul[15Ul[4R6"U[8X7X`R2UR.5Ulg)aEnter `cm` in a persistent task on the caller's event loop, capturing its context variables. Args: cm: The async streaming context manager to run on the caller's event loop. async_alternative: How to name the async counterpart in error messages (e.g. `run_stream`). zdCannot use a synchronous streaming method from within an async context or a running event loop; use z instead.N)r3rXr;rget_event_loopr create_futurer9r0r:r$r2rEr rBr?r-_loop _owner_task_exit_requested_caller_context _run_contextr _owner_thread_idset _pump_tasksweakreffinalizer[ _finalizer) selfr*async_alternativer@caller_contextr+r,rGr- run_contexts r__init__SyncStreamBridge.__init__s|   $ $ &##4"5Y@    $$&%;?;M;M;O484F4F4H%%&;B&XY  "&"9"9'"B FK ??$$!!#-('9)((-( ) )(    %--' ) 47E!** .$NDTDTVZVkVk sH& 33B%%>E# C8/ E8 D ED2) E2 E <Ec[5nURHQnURUnX R;dURUULdM5URURU5 MS U$)zGMerge run-owned context changes into the sync caller's current context.)r rmrlrunro)rtcontextvarrs r _task_contextSyncStreamBridge._task_contextsa.$$C%%c*E...$2F2Fs2KSX2X CGGU+%rc[5UR:wa [S5e[R"5 [S5e![a gf=f)NzPA synchronous stream must be used and closed on the thread where it was created.zMA synchronous stream cannot be used or closed while an event loop is running.)r rnr;r3rX)rts r_check_owner_thread$SyncStreamBridge._check_owner_threadsQ ;$// /qr r   $ $ &jkk   sA AAcUR5 URR5b8[URUR UR URU5 gg)aQExit the stream context manager, at most once. `detach()` disarms the finalizer (returning true iff it was still live), guarding against a double shutdown from the owning wrapper's `__exit__`, a Ctrl-C teardown, and a later GC. The `__exit__` arguments are passed to the stream context manager so it can tear the stream down correctly. N)rrsdetachrJrirjrkrp)rtr/s rshutdownSyncStreamBridge.shutdownsO   " ?? ! ! # / 4::t'7'79M9MtO_O_aj k 0rcUR5R[RXRS9nUR 5(aUR 5$URRU5$![al UR 5(dUUR5 [[5 URRU5 SSS5 e!,(df  e=fef=f)zZRun `awaitable` on the bridge's event loop and clean up its task if the caller interrupts.)r@N) r~r{r3 ensure_futurerir2r?r:r$rEr )rt awaitabler6s r_runSyncStreamBridge._runs!!#''(=(=yzz'Z 99;;;;=  ::006 6 99;; m,JJ11$7- -, E  s$A88>C.6C C. C) % C.cP# U"U6n[U5(a UIShvN $U$N7frR)r)rtfuncargsr?s r_callSyncStreamBridge._call#s+t  <   s &$&c2URR(d [S5eUR5 UR UR "U/UQ765$![ [4a,nUR[U5X3R45 eSnAff=f)aoRun `func` on the bridge's event loop, tearing the run down if the caller is interrupted. Without this, a `KeyboardInterrupt` (Ctrl-C) or `SystemExit` landing while we're blocked on the event loop would unwind the caller while leaving the async code's pending tasks and open sockets until garbage collection. See https://github.com/pydantic/pydantic-ai/issues/5975. *This synchronous stream is already closed.N) rsaliver;rrrr"r#rr(r))rtrrr.s rcallSyncStreamBridge.call)s~$$KL L   " 99TZZ4t45 5!:.  MM49c+<+<= >  s!AB*'BBc# UIShvN U"5nUShvN nURU5IShvN M$N2N!N [U[5(aUR5IShvN O4![U[5(aUR5IShvN ff=fSSS5IShvN g!,IShvN (df  g=f7f)zDrive `make_aiter()` to completion in one task, forwarding items to `send_stream`. Running the whole `async for` in one task keeps the source iterator's cancel scopes (e.g. `group_by_temporal`'s) from being entered and exited in different tasks. N)send isinstanceraclose) make_aiter send_streamaiteritems r_pump_to_stream SyncStreamBridge._pump_to_stream9s;LE )"'1$%**4000 10#( e^44,,.((e^44,,.((5;;;;;sC<CB8A2A>AA2AA2CAA2A2(B8+A.,B82)B#B B##B8& C1B42C8C>C ? C Cc#>^^^^ ^ # TRR(d [S5eTR5 [R [ "SS9unm TR5RTRRTRX55mTRm T RT5 S U 4SjjnTRU5 S UU 4SjjmS UUU U4SjjnSnTR[ T 5n["R$"U5(dOUv M8TR'T5 U(d]TR5 T"5 [+[,5 TR'T5 S S S 5 TRR/T5 g g ![(a TR5 e![a U"5 SnU(dTR5 O![a U"5 ef=fT"5 [+[,5 TR'T5 S S S 5 O!,(df  O=fTRR/T5 g g f=ff=f![a U"5 ef=f!,(df  GN =f!U(dTR5 O![a U"5 ef=fT"5 [+[,5 TR'T5 S S S 5 O!,(df  O=fTRR/T5 ff=f7f) zVSynchronously iterate the items produced by `make_aiter()` on the bridge's event loop.rr)max_buffer_sizec>TRU5 [[5 UR5 SSS5 g!,(df  g=frR)discardr r$r>)r6rHs r discard_pump2SyncStreamBridge.stream_sync..discard_pumpWs1   t $-( )((s < A cr>TR5 TR5(dTR5 ggrR)closer2rE) pump_taskrasr cancel_pump1SyncStreamBridge.stream_sync..cancel_pump`s-  ">>##  "$rc>TR5(aTR5 g[[5 TRR T5 SSS5 g!,(df  g=frR)r2rr r;rirY)rrrartsrdefer_pump_cleanup8SyncStreamBridge.stream_sync..defer_pump_cleanupesD~~$$&l+JJ33K@,++s A A*FTNr6asyncio.Task[None]rVrWrU)rsrr;rr^create_memory_object_streamrr~r{rir9rrpaddadd_done_callbackrrbris_setr GeneratorExitr r$r) rtrrrrcleanup_deferredreceivedrrrHras ` @@@@r stream_syncSyncStreamBridge.stream_syncMs$$KL L   "&+&G&G&J[\&]# ^&&(,,TZZ-C-CTEYEYZdErs %% y! ! ##L1 # #  A A!" 499\>B}}X..  IIi $,,. m,IIi(-  ((3$  ((*    "$#' $,,.#'(  m,IIi(-,,  ((3$    $'( -,$,,.#'(  m,IIi(-,,  ((3$sC$L,A F7L?I#L%I97%L I 'F87I 8I I  J LG&%L&G99LH-$ L- H;7$LII  J #I66L9 JL LJ%$L%J88LK,# L, K: 6$LL) rlrkrsrirjrnrprmr-N)r*$AbstractAsyncContextManager[StreamT]rustrrVrW)rVr rU)rM)r/ _ExitInforVrW)rz Awaitable[T]rVr)rz/Callable[[Unpack[_PosArgsT]], Awaitable[T] | T]rzUnpack[_PosArgsT]rVr)rCallable[[], AsyncIterator[T]]rz.anyio.streams.memory.MemoryObjectSendStream[T]rVrW)rrrVz Iterator[T])__name__ __module__ __qualname____firstlineno____doc____annotations__rxr~rrrrr staticmethodrr__static_attributes__rrrdrdsi  O:. `l l   )2)Ao) ))&D4rrd)rzT | Awaitable[T]rVzTypeIs[Awaitable[T]])r*rr+z'asyncio.Future[tuple[StreamT, Context]]r,asyncio.Future[_ExitInfo]rVrWr)r@asyncio.AbstractEventLoopr6rrVrW) r@rrGrr,rrHset[asyncio.Task[None]]r/rrVrW)rGrr,rrHrrVrW) r@rrGrr,rrHrrZintrVrW)raz1anyio.streams.memory.MemoryObjectReceiveStream[T]rVzT | _utils.Unset)1r __future__rr3rrqcollections.abcrrrrr contextlibr r contextvarsr r threadingr typesrtypingrr^anyio.streams.memorytyping_extensionsrrrrrrrrrDr(r$rrr0r7rBrJrOr[rbrdrrrrsr #XX<- CC CL )   % $}%,md.BMTXDXX Y & ',' 4'.' '>.. #."...( .  .  .*"-( (0 #0"0.0( 0  0  06[4ww'[4r